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

Merge branch 'master' into update-grpc

Browse Source 
# Conflicts:
#	gradle/libs.versions.toml
This commit is contained in:
Maksim Fomenkov
2022-11-14 18:08:49 +03:00
parent 198db59818 809844d452
commit 9fc701dfe4
55 changed files with 793 additions and 988 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
docs/04-upstream-config.adoc
View File

@@ -51,6 +51,7 @@ cluster:
min-peers: 2
upstreams:
- id: us-nodes
node-id: 1
chain: auto
connection:
grpc:
@@ -61,6 +62,7 @@ cluster:
certificate: client.crt
key: client.p8.key
- id: infura-eth
node-id: 2
chain: ethereum
role: fallback
labels:
@@ -80,6 +82,7 @@ cluster:
username: ${INFURA_USER}
password: ${INFURA_PASSWD}
- id: ethereum-pos
node-id: 3
chain: ropsten
connection:
ethereum-pos:
@@ -110,6 +113,12 @@ In the example above we have:
** label `[provider: infura]` is set for that particular upstream, which can be selected during a request.For example for some requests you may want to use nodes with that label only, i.e. _"send that tx to infura nodes only"_, or _"read only from archive node, with label [archive: true]"_
** upstream validation (peers, sync status, etc) is disabled for that particular upstream
=== Nodes
`[node-id: 1]` is numeric node identifier defined in a range [1..255] and used to forward
`eth_getFilterChanges` request to the node where one of `eth_newFilter`, `eth_newBlockFilter` or `eth_newPendingTransactionFilter` methods was executed (because filter is s stateful method).
*It's kindly recommended* to strictly associate _node-id_ parameter with a physical node and keep it during any configuration changes
=== Roles and Fallback upstream
By default, the Dshackle connects to each upstream in a Round-Robin basis, i.e. sequentially one by one.
@@ -206,6 +215,10 @@ Dshackle currently supports
- `eth_getUncleByBlockNumberAndIndex`
- `eth_feeHistory`
- `eth_getLogs`
- `eth_getFilterChanges`
- `eth_newFilter`
- `eth_newBlockFilter`
- `eth_newPendingTransactionFilter`
.Plus following methods are answered directly by Dshackle
- `net_version`

23
dshackle-cli/README.md
View File

@@ -1,21 +1,36 @@
# DSHACKLE CLI
A CLI tool to verify the [dshacle](https://github.com/emeraldpay/dshackle) installation.
A CLI tool to verify the [dshackle](https://github.com/emeraldpay/dshackle) installation.
## Usage
```
npx dchackle-cli-tool [-p|--print] <dshackle instance url>
npx dshackle-cli-tool [-p|--print][--ca][--cert][--key] <dshackle instance url>
```
### Options
-p | --print - will print the describe response as is
```-p | --print``` will print the describe response as is
### TLS
```--ca``` the root certificate data
```--cert``` the client certificate key chain, if available
```--key``` the client certificate private key, if available
## Example
```
> dchackle-cli-tool localhost:2449
> dshackle-cli-tool localhost:2449
Connecting to: localhost:2449...
Connected to localhost:2449
CHAIN_ETHEREUM -> AVAIL_OK
CHAIN_KOVAN -> AVAIL_OK
```
With TLS
```
>dshackle-cli % dshackle-cli-tool --ca ./../out/ca.myhost.dev.crt --cert ./../out/client_1.crt --key ./../out/client_1.key 127.0.0.1:2450
Using TLS
Connecting to: 127.0.0.1:2450...
Connected to 127.0.0.1:2450
CHAIN_ETHEREUM -> AVAIL_OK
```

2
dshackle-cli/grpc

Submodule dshackle-cli/grpc updated: 4ed721cb90...4060e53997

879
dshackle-cli/package-lock.json generated
View File

File diff suppressed because it is too large Load Diff

5
dshackle-cli/package.json
View File

@@ -1,6 +1,6 @@
{
"name": "dshackle-cli-tool",
"version": "1.0.4",
"version": "1.0.7",
"description": "",
"main": "src/index.js",
"scripts": {
@@ -18,8 +18,7 @@
"@grpc/proto-loader": "^0.7.2",
"arg": "^5.0.2",
"cli-color": "^2.0.3",
"esm": "^3.2.25",
"inquirer": "^9.1.1"
"esm": "^3.2.25"
},
"files": [
"bin/",

121
dshackle-cli/src/cli.js
View File

@@ -1,47 +1,112 @@
import {describe} from "./grpc-clent";
import {describe, connect, nativeCall} from "./grpc-clent";
import arg from 'arg';
import clc from "cli-color";
import util from "util";
const chains = {
CHAIN_BSC: 1006,
CHAIN_OPTIMISM: 1005,
CHAIN_ARBITRUM: 1004,
CHAIN_MATIC: 1002,
CHAIN_ETHEREUM: 100
}
export function cli(args) {
let opts = parseArgumentsIntoOptions(args);
if (!opts.url) {
console.log("Err: URL not specified!")
console.log("Err: URL not specified!!!")
return
}
describe(opts.url, (error, response) => {
if (error) {
console.error(clc.red('Connection to ' + opts.url + ' failed! [' + error.message + ']'));
} else {
console.error(clc.green('Connected to ', opts.url));
if (opts.print) {
console.log(util.inspect(response, false, null, true /* enable colors */));
} else {
response.chains.forEach(function (item) {
var state = item.status.availability;
switch (state) {
case 'AVAIL_OK':
state = clc.green(state);
break
case 'AVAIL_UNKNOWN':
case 'AVAIL_UNAVAILABLE':
state = clc.red(state);
break
default:
state = clc.yellow(state);
}
console.log(item.status.chain + ' -> ' + clc.bold(state))
})
const client = connect(opts.url, opts.ca, opts.cert, opts.key)
describe(client, (error, response) => {
if (error) {
console.error(clc.red('Connection to ' + opts.url + ' failed: ' + error.message));
return
}
console.log(clc.green('Connected to ', opts.url));
if (opts.print) {
console.log(util.inspect(response, false, null, true /* enable colors */));
}
processDescribe(client, response, opts.testRun)
})
}
function processDescribe(client, response, testRun) {
let promises = []
let statuses = new Map()
response.chains.forEach((item) => {
const state = item.status.availability
const chain = item.status.chain
let status = {
state: 'AVAIL_UNKNOWN',
grpc: clc.yellow('UNKNOWN'),
failed: false
}
switch (state) {
case 'AVAIL_OK':
status.state = clc.green(state);
break
case 'AVAIL_UNKNOWN':
case 'AVAIL_UNAVAILABLE':
status.state = clc.red(state);
break
default:
status.state = clc.yellow(state);
}
if (state === 'AVAIL_OK') {
promises.push(nativeCall(client, chains[chain], chain))
} else {
status.failed = true
}
statuses.set(chain, status)
})
Promise.all(promises).then(responses => {
responses.forEach((resp) => {
let status = statuses.get(resp.chain)
if (resp.error) {
status.failed = true
status.grpc = clc.red(resp.error.message)
} else {
if (resp.payload.succeed) {
status.grpc = clc.green('OK')
} else {
status.grpc = clc.red('FAILED')
status.failed = true
}
}
})
let hasError = false
statuses.forEach((status, chain) => {
printState(chain, status)
if (status.failed) {
hasError = true
}
})
if (hasError && testRun) {
process.exit(1)
}
})
}
function printState(chain, status) {
console.log(chain + ' -> ' + "state: " + clc.bold(status.state) + " gRPC: " + clc.bold(status.grpc))
}
function parseArgumentsIntoOptions(rawArgs) {
const args = arg(
{
'--print': Boolean,
'--test-run': Boolean,
'--ca': String,
'--cert': String,
'--key': String,
'-p': '--print'
},
{
@@ -50,7 +115,11 @@ function parseArgumentsIntoOptions(rawArgs) {
);
return {
print: args['--print'] || false,
url: args._[0]
testRun: args['--test-run'] || false,
url: args._[0],
ca: args['--ca'],
cert: args['--cert'],
key: args['--key']
};
}

54
dshackle-cli/src/grpc-clent.js
View File

@@ -1,6 +1,7 @@
const grpc = require("@grpc/grpc-js");
const path = require('path')
const protoLoader = require("@grpc/proto-loader");
const fs = require('fs');
const PROTO_PATH = path.join(__dirname, "../grpc/proto/blockchain.proto");
@@ -15,11 +16,54 @@ const options = {
const packageDefinition = protoLoader.loadSync(PROTO_PATH, options);
const emerald = grpc.loadPackageDefinition(packageDefinition).emerald
export function describe(url, handler) {
const client = new emerald.Blockchain(
url,
grpc.credentials.createInsecure()
);
var id = 100
export function connect(url, ca, cert, key) {
let credentials = grpc.credentials.createInsecure()
if (ca || cert || key) {
console.log("Using TLS")
credentials = grpc.credentials.createSsl(
ca ? fs.readFileSync(ca) : null,
key ? fs.readFileSync(key) : null,
cert ? fs.readFileSync(cert) : null
);
}
console.log('Connecting to: ' + url + '...')
return new emerald.Blockchain(
url,
credentials
);
}
export function describe(client, handler) {
client.Describe({}, handler);
}
export function nativeCall(client, chainCode, chain) {
return new Promise((resolve, reject) => {
const call = client.NativeCall({
chain: chainCode,
items: [{
id: id++,
method: "eth_getBalance",
payload: "WyIweDhEOTc2ODlDOTgxODg5MkI3MDBlMjdGMzE2Y2MzRTQxZTE3ZkJlYjkiLCAibGF0ZXN0Il0="
}],
quorum: 1,
min_availability: 0
})
call.on('data', (item) => {
resolve(toResult(chain, item, null))
})
call.on('end', () => resolve(toResult(chain, null, null)))
call.on('error', (e) => reject(toResult(chain, null, e)))
})
}
function toResult(chain, obj, err) {
return {
chain: chain,
payload: obj,
error: err
}
}

2
emerald-java-client

Submodule emerald-java-client updated: 99c7f33060...35196f35f3

2
gradle/libs.versions.toml
View File

@@ -5,7 +5,7 @@ groovy = "2.5.14"
protoc = "3.21.7"
slf4j = "1.7.32"
jackson = "2.11.0"
grpc = "1.38.0"
grpc = "1.49.2"
reactive-grpc = "1.2.0"
spring-boot = "2.5.6"
spring-security = "5.5.3"

10
src/main/kotlin/io/emeraldpay/dshackle/Global.kt
View File

@@ -44,8 +44,14 @@ class Global {
"ethereum" to Chain.ETHEREUM,
"ethereum-classic" to Chain.ETHEREUM_CLASSIC,
"eth" to Chain.ETHEREUM,
"polygon" to Chain.MATIC,
"matic" to Chain.MATIC,
"polygon" to Chain.POLYGON,
"matic" to Chain.POLYGON,
"arbitrum" to Chain.ARBITRUM,
"arb" to Chain.ARBITRUM,
"optimism" to Chain.OPTIMISM,
"binance" to Chain.BSC,
"bsc" to Chain.BSC,
"bnb-smart-chain" to Chain.BSC,
"etc" to Chain.ETHEREUM_CLASSIC,
"morden" to Chain.TESTNET_MORDEN,
"kovan" to Chain.TESTNET_KOVAN,

2
src/main/kotlin/io/emeraldpay/dshackle/config/TokensConfig.kt
View File

@@ -41,7 +41,7 @@ class TokensConfig(
type == null -> type
address.isNullOrBlank() -> "address"
blockchain != null &&
(BlockchainType.from(blockchain!!) == BlockchainType.ETHEREUM_POS || BlockchainType.from(blockchain!!) == BlockchainType.ETHEREUM) &&
(BlockchainType.from(blockchain!!) == BlockchainType.EVM_POS || BlockchainType.from(blockchain!!) == BlockchainType.EVM_POW) &&
!Address.isValidAddress(address) -> "address"
else -> null
}

1
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfig.kt
View File

@@ -69,6 +69,7 @@ open class UpstreamsConfig {
class Upstream<T : UpstreamConnection> {
var id: String? = null
var nodeId: Int? = null
var chain: String? = null
var options: Options? = null
var isEnabled = true

14
src/main/kotlin/io/emeraldpay/dshackle/config/UpstreamsConfigReader.kt
View File

@@ -31,6 +31,7 @@ class UpstreamsConfigReader(
private val log = LoggerFactory.getLogger(UpstreamsConfigReader::class.java)
private val authConfigReader = AuthConfigReader()
private val knownNodeIds: MutableSet<Int> = HashSet()
fun read(input: InputStream): UpstreamsConfig? {
val configNode = readNode(input)
@@ -236,11 +237,22 @@ class UpstreamsConfigReader(
log.warn("Invalid id: $id")
return false
}
return true
return upstream.nodeId?.let {
if (it !in 1..255) {
log.warn("Invalid node-id: $it. Must be in range [1, 255].")
false
} else if (!knownNodeIds.add(it)) {
log.warn("Duplicated node-id: $it. Must be in unique.")
false
} else {
true
}
} ?: true
}
internal fun readUpstreamCommon(upNode: MappingNode, upstream: UpstreamsConfig.Upstream<*>) {
upstream.id = getValueAsString(upNode, "id")
upstream.nodeId = getValueAsInt(upNode, "node-id")
upstream.options = tryReadOptions(upNode)
upstream.methods = tryReadMethods(upNode)
getValueAsBool(upNode, "enabled")?.let {

6
src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt
View File

@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import java.util.concurrent.ConcurrentLinkedQueue
open class AlwaysQuorum : CallQuorum {
@@ -28,6 +29,7 @@ open class AlwaysQuorum : CallQuorum {
private var result: ByteArray? = null
private var rpcError: JsonRpcError? = null
private var sig: ResponseSigner.Signature? = null
private val resolvers: MutableCollection<Upstream> = ConcurrentLinkedQueue()
override fun init(head: Head) {
}
@@ -48,6 +50,7 @@ open class AlwaysQuorum : CallQuorum {
result = response
resolved = true
sig = signature
resolvers.add(upstream)
return true
}
@@ -64,6 +67,9 @@ open class AlwaysQuorum : CallQuorum {
return rpcError
}
override fun getResolvedBy(): List<Upstream> =
resolvers.toList()
override fun toString(): String {
return "Quorum: Accept Any"
}

1
src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt
View File

@@ -34,4 +34,5 @@ interface CallQuorum {
fun getSignature(): ResponseSigner.Signature?
fun getResult(): ByteArray?
fun getError(): JsonRpcError?
fun getResolvedBy(): Collection<Upstream>
}

5
src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt
View File

@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.atomic.AtomicReference
/**
@@ -34,6 +35,7 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
private val failed = AtomicReference(false)
private var rpcError: JsonRpcError? = null
private var sig: ResponseSigner.Signature? = null
private val resolvers: MutableCollection<Upstream> = ConcurrentLinkedQueue()
override fun init(head: Head) {
}
@@ -51,6 +53,7 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
if (!lagging) {
result.set(response)
sig = signature
resolvers.add(upstream)
return true
}
return false
@@ -75,6 +78,8 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
return rpcError
}
override fun getResolvedBy(): Collection<Upstream> =
resolvers.toList()
override fun toString(): String {
return "Quorum: late <= $maxLag blocks"
}

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

@@ -117,9 +117,9 @@ class QuorumRpcReader(
return Function { quorumResult ->
quorumResult
.filter { it.isResolved() } // return nothing if not resolved
.map {
.map { quorum ->
// TODO find actual quorum number
QuorumRpcReader.Result(it.getResult()!!, it.getSignature(), 1)
Result(quorum.getResult()!!, quorum.getSignature(), 1, quorum.getResolvedBy().map { it.nodeId() })
}
.switchIfEmpty(defaultResult)
}
@@ -197,6 +197,7 @@ class QuorumRpcReader(
class Result(
val value: ByteArray,
val signature: ResponseSigner.Signature?,
val quorum: Int
val quorum: Int,
val resolvers: Collection<Byte>
)
}

6
src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt
View File

@@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.emeraldpay.etherjar.rpc.RpcException
import org.slf4j.LoggerFactory
import java.util.concurrent.ConcurrentLinkedQueue
abstract class ValueAwareQuorum<T>(
val clazz: Class<T>
@@ -30,6 +31,7 @@ abstract class ValueAwareQuorum<T>(
private val log = LoggerFactory.getLogger(ValueAwareQuorum::class.java)
private var rpcError: JsonRpcError? = null
private val resolvers: MutableCollection<Upstream> = ConcurrentLinkedQueue()
fun extractValue(response: ByteArray, clazz: Class<T>): T? {
return Global.objectMapper.readValue(response.inputStream(), clazz)
@@ -39,6 +41,7 @@ abstract class ValueAwareQuorum<T>(
try {
val value = extractValue(response, clazz)
recordValue(response, value, signature, upstream)
resolvers.add(upstream)
} catch (e: RpcException) {
recordError(response, e.rpcMessage, signature, upstream)
} catch (e: Exception) {
@@ -59,4 +62,7 @@ abstract class ValueAwareQuorum<T>(
override fun getError(): JsonRpcError? {
return rpcError
}
override fun getResolvedBy(): Collection<Upstream> =
resolvers.toList()
}

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

@@ -31,6 +31,7 @@ import io.emeraldpay.dshackle.upstream.Multistream
import io.emeraldpay.dshackle.upstream.MultistreamHolder
import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.calls.EthereumCallSelector
import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
@@ -66,14 +67,16 @@ open class NativeCall(
private val ethereumCallSelectors = EnumMap<Chain, EthereumCallSelector>(Chain::class.java)
init {
val casting = mapOf(
BlockchainType.EVM_POS to EthereumPosMultiStream::class.java,
BlockchainType.EVM_POW to EthereumMultistream::class.java,
)
multistreamHolder.observeChains().subscribe { chain ->
if ((BlockchainType.from(chain) == BlockchainType.ETHEREUM_POS || BlockchainType.from(chain) == BlockchainType.ETHEREUM) && !ethereumCallSelectors.containsKey(
chain
)
) {
casting[BlockchainType.from(chain)]?.let { cast ->
multistreamHolder.getUpstream(chain)?.let { up ->
val reader = up.cast(EthereumPosMultiStream::class.java).getReader()
ethereumCallSelectors[chain] = EthereumCallSelector(reader.heightByHash())
val reader = up.cast(cast).getReader()
ethereumCallSelectors.putIfAbsent(chain, EthereumCallSelector(reader.heightByHash()))
}
}
}
@@ -102,7 +105,8 @@ open class NativeCall(
}
fun parseParams(it: ValidCallContext<RawCallDetails>): ValidCallContext<ParsedCallDetails> {
val params = extractParams(it.payload.params)
val rawParams = extractParams(it.payload.params)
val params = it.requestDecorator.processRequest(rawParams)
return it.withPayload(ParsedCallDetails(it.payload.method, params))
}
@@ -212,7 +216,7 @@ open class NativeCall(
}
// for ethereum the actual block needed for the call may be specified in the call parameters
val callSpecificMatcher: Mono<Selector.Matcher> =
if (BlockchainType.from(upstream.chain) == BlockchainType.ETHEREUM_POS || BlockchainType.from(upstream.chain) == BlockchainType.ETHEREUM) {
if (BlockchainType.from(upstream.chain) == BlockchainType.EVM_POS || BlockchainType.from(upstream.chain) == BlockchainType.EVM_POW) {
ethereumCallSelectors[chain]?.getMatcher(method, params, upstream.getHead())
} else {
null
@@ -234,17 +238,31 @@ open class NativeCall(
matcher.withMatcher(heightMatcher)
}
val nonce = requestItem.nonce.let { if (it == 0L) null else it }
val requestDecorator = getRequestDecorator(requestItem.method)
val resultDecorator = getResultDecorator(requestItem.method)
ValidCallContext(
requestItem.id,
nonce,
upstream,
matcher.build(),
callQuorum,
RawCallDetails(method, params)
RawCallDetails(method, params),
requestDecorator,
resultDecorator
)
}
}
private fun getRequestDecorator(method: String): RequestDecorator =
if (method == "eth_getFilterChanges" || method == "eth_uninstallFilter")
GetFilterUpdatesDecorator()
else
NoneRequestDecorator()
private fun getResultDecorator(method: String): ResultDecorator =
if (CreateFilterDecorator.createFilterMethods.contains(method)) CreateFilterDecorator() else NoneResultDecorator()
fun fetch(ctx: ValidCallContext<ParsedCallDetails>): Mono<CallResult> {
return ctx.upstream.getRoutedApi(ctx.matcher)
.flatMap { api ->
@@ -283,7 +301,8 @@ open class NativeCall(
return reader
.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce))
.map {
CallResult(ctx.id, ctx.nonce, it.value, null, it.signature)
val bytes = ctx.resultDecorator.processResult(it)
CallResult(ctx.id, ctx.nonce, bytes, null, it.signature)
}
.onErrorResume { t ->
val failure = when (t) {
@@ -350,14 +369,71 @@ open class NativeCall(
fun getError(): CallError
}
interface ResultDecorator {
fun processResult(result: QuorumRpcReader.Result): ByteArray
}
open class NoneResultDecorator : ResultDecorator {
override fun processResult(result: QuorumRpcReader.Result): ByteArray = result.value
}
open class CreateFilterDecorator : ResultDecorator {
companion object {
const val quoteCode = '"'.code.toByte()
val createFilterMethods = listOf(
"eth_newFilter",
"eth_newBlockFilter",
"eth_newPendingTransactionFilter"
)
}
override fun processResult(result: QuorumRpcReader.Result): ByteArray {
val bytes = result.value
if (bytes.last() == quoteCode) {
val suffix = result.resolvers.first().toUByte().toString(16).padStart(2, padChar = '0').toByteArray()
bytes[bytes.lastIndex] = suffix.first()
return bytes + suffix.last() + quoteCode
}
return bytes
}
}
interface RequestDecorator {
fun processRequest(request: List<Any>): List<Any>
}
open class NoneRequestDecorator : RequestDecorator {
override fun processRequest(request: List<Any>): List<Any> = request
}
open class GetFilterUpdatesDecorator : RequestDecorator {
override fun processRequest(request: List<Any>): List<Any> {
val filterId = request.first().toString()
val sanitized = filterId.substring(0, filterId.lastIndex - 1)
return listOf(sanitized)
}
}
open class ValidCallContext<T>(
val id: Int,
val nonce: Long?,
val upstream: Multistream,
val matcher: Selector.Matcher,
val callQuorum: CallQuorum,
val payload: T
val payload: T,
val requestDecorator: RequestDecorator,
val resultDecorator: ResultDecorator
) : CallContext {
constructor(
id: Int,
nonce: Long?,
upstream: Multistream,
matcher: Selector.Matcher,
callQuorum: CallQuorum,
payload: T
) : this(id, nonce, upstream, matcher, callQuorum, payload, NoneRequestDecorator(), NoneResultDecorator())
override fun isValid(): Boolean {
return true
}
@@ -371,7 +447,7 @@ open class NativeCall(
}
fun <X> withPayload(payload: X): ValidCallContext<X> {
return ValidCallContext(id, nonce, upstream, matcher, callQuorum, payload)
return ValidCallContext(id, nonce, upstream, matcher, callQuorum, payload, requestDecorator, resultDecorator)
}
fun getApis(): ApiSource {

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

@@ -57,7 +57,7 @@ open class NativeSubscribe(
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) {
if (BlockchainType.from(chain) != BlockchainType.EVM_POS && BlockchainType.from(chain) != BlockchainType.EVM_POW) {
return Mono.error(UnsupportedOperationException("Native subscribe is not supported for ${chain.chainCode}"))
}

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

@@ -70,7 +70,7 @@ class TrackERC20Address(
override fun isSupported(chain: Chain, asset: String): Boolean {
return tokens.containsKey(TokenId(chain, asset.lowercase(Locale.getDefault()))) &&
(BlockchainType.from(chain) == BlockchainType.ETHEREUM_POS || BlockchainType.from(chain) == BlockchainType.ETHEREUM) && multistreamHolder.isAvailable(chain)
(BlockchainType.from(chain) == BlockchainType.EVM_POS || BlockchainType.from(chain) == BlockchainType.EVM_POW) && multistreamHolder.isAvailable(chain)
}
override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {

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

@@ -43,7 +43,7 @@ class TrackEthereumAddress(
override fun isSupported(chain: Chain, asset: String): Boolean {
return asset == "ether" &&
(BlockchainType.from(chain) == BlockchainType.ETHEREUM_POS || BlockchainType.from(chain) == BlockchainType.ETHEREUM) && multistreamHolder.isAvailable(chain)
(BlockchainType.from(chain) == BlockchainType.EVM_POS || BlockchainType.from(chain) == BlockchainType.EVM_POW) && multistreamHolder.isAvailable(chain)
}
override fun getBalance(request: BlockchainOuterClass.BalanceRequest): Flux<BlockchainOuterClass.AddressBalance> {

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

@@ -62,7 +62,7 @@ class TrackEthereumTx(
private val log = LoggerFactory.getLogger(TrackEthereumTx::class.java)
override fun isSupported(chain: Chain): Boolean {
return (BlockchainType.from(chain) == BlockchainType.ETHEREUM_POS || BlockchainType.from(chain) == BlockchainType.ETHEREUM) && multistreamHolder.isAvailable(chain)
return (BlockchainType.from(chain) == BlockchainType.EVM_POS || BlockchainType.from(chain) == BlockchainType.EVM_POW) && multistreamHolder.isAvailable(chain)
}
override fun subscribe(request: BlockchainOuterClass.TxStatusRequest): Flux<BlockchainOuterClass.TxStatus> {

98
src/main/kotlin/io/emeraldpay/dshackle/startup/ConfiguredUpstreams.kt
View File

@@ -16,6 +16,7 @@
*/
package io.emeraldpay.dshackle.startup
import com.google.common.annotations.VisibleForTesting
import io.emeraldpay.dshackle.FileResolver
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.config.UpstreamsConfig
@@ -53,7 +54,9 @@ import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Repository
import java.net.URI
import java.util.concurrent.atomic.AtomicInteger
import java.util.function.Function
import javax.annotation.PostConstruct
import kotlin.math.abs
@Repository
open class ConfiguredUpstreams(
@@ -65,6 +68,8 @@ open class ConfiguredUpstreams(
private val log = LoggerFactory.getLogger(ConfiguredUpstreams::class.java)
private var seq = AtomicInteger(0)
private val hashes: MutableMap<Byte, Boolean> = HashMap()
@PostConstruct
fun start() {
log.debug("Starting upstreams")
@@ -77,7 +82,7 @@ open class ConfiguredUpstreams(
log.debug("Start upstream ${up.id}")
if (up.connection is UpstreamsConfig.GrpcConnection) {
val options = up.options ?: UpstreamsConfig.Options()
buildGrpcUpstream(up.cast(UpstreamsConfig.GrpcConnection::class.java), options)
buildGrpcUpstream(up.nodeId, up.cast(UpstreamsConfig.GrpcConnection::class.java), options)
} else {
val chain = Global.chainById(up.chain)
if (chain == Chain.UNSPECIFIED) {
@@ -87,15 +92,23 @@ open class ConfiguredUpstreams(
val options = (up.options ?: UpstreamsConfig.Options())
.merge(defaultOptions[chain] ?: UpstreamsConfig.Options.getDefaults())
val upstream = when (BlockchainType.from(chain)) {
BlockchainType.ETHEREUM -> {
buildEthereumUpstream(up.cast(UpstreamsConfig.EthereumConnection::class.java), chain, options)
BlockchainType.EVM_POW -> {
buildEthereumUpstream(up.nodeId, up.cast(UpstreamsConfig.EthereumConnection::class.java), chain, options)
}
BlockchainType.BITCOIN -> {
buildBitcoinUpstream(up.cast(UpstreamsConfig.BitcoinConnection::class.java), chain, options)
}
BlockchainType.ETHEREUM_POS -> {
buildEthereumPosUpstream(up.cast(UpstreamsConfig.EthereumPosConnection::class.java), chain, options)
BlockchainType.EVM_POS -> {
buildEthereumPosUpstream(
up.nodeId,
up.cast(UpstreamsConfig.EthereumPosConnection::class.java),
chain,
options
)
}
else -> {
log.error("Chain is unsupported: ${up.chain}")
return@forEach
@@ -110,7 +123,7 @@ open class ConfiguredUpstreams(
fun hasMatchingUpstream(chain: Chain, matcher: Selector.LabelSelectorMatcher): Boolean =
config.upstreams.any { up ->
up.chain?.equals(chain.chainName, ignoreCase = true) ?: true && matcher.matches(up.labels)
(up.chain?.let { Global.chainById(it) == chain } ?: true) && matcher.matches(up.labels)
}
private fun buildDefaultOptions(config: UpstreamsConfig): HashMap<Chain, UpstreamsConfig.Options> {
@@ -156,6 +169,7 @@ open class ConfiguredUpstreams(
}
private fun buildEthereumPosUpstream(
nodeId: Int?,
config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumPosConnection>,
chain: Chain,
options: UpstreamsConfig.Options
@@ -167,13 +181,26 @@ open class ConfiguredUpstreams(
return null
}
val urls = ArrayList<URI>()
val connectorFactory = buildEthereumConnectorFactory(config.id!!, execution, chain, urls, NoChoiceWithPriorityForkChoice(conn.upstreamRating), BlockValidator.ALWAYS_VALID)
val connectorFactory = buildEthereumConnectorFactory(
config.id!!,
execution,
chain,
urls,
NoChoiceWithPriorityForkChoice(conn.upstreamRating),
BlockValidator.ALWAYS_VALID
)
val methods = buildMethods(config, chain)
if (connectorFactory == null) {
return null
}
val hashUrl = conn.execution!!.let {
if (it.preferHttp) it.rpc?.url ?: it.ws?.url else it.ws?.url ?: it.rpc?.url
}
val hash = getHash(nodeId, hashUrl!!)
val upstream = EthereumPosRpcUpstream(
config.id!!,
hash,
chain,
options, config.role,
methods,
@@ -227,6 +254,7 @@ open class ConfiguredUpstreams(
}
private fun buildEthereumUpstream(
nodeId: Int?,
config: UpstreamsConfig.Upstream<UpstreamsConfig.EthereumConnection>,
chain: Chain,
options: UpstreamsConfig.Options
@@ -236,12 +264,22 @@ open class ConfiguredUpstreams(
val urls = ArrayList<URI>()
val methods = buildMethods(config, chain)
val connectorFactory = buildEthereumConnectorFactory(config.id!!, conn, chain, urls, MostWorkForkChoice(), EthereumBlockValidator())
val connectorFactory = buildEthereumConnectorFactory(
config.id!!,
conn,
chain,
urls,
MostWorkForkChoice(),
EthereumBlockValidator()
)
if (connectorFactory == null) {
return null
}
val hashUrl = if (conn.preferHttp) conn.rpc?.url ?: conn.ws?.url else conn.ws?.url ?: conn.rpc?.url
val upstream = EthereumRpcUpstream(
config.id!!,
getHash(nodeId, hashUrl!!),
chain,
options, config.role,
methods,
@@ -253,12 +291,15 @@ open class ConfiguredUpstreams(
}
private fun buildGrpcUpstream(
nodeId: Int?,
config: UpstreamsConfig.Upstream<UpstreamsConfig.GrpcConnection>,
options: UpstreamsConfig.Options
) {
val endpoint = config.connection!!
val hash = getHash(nodeId, "${endpoint.host}:${endpoint.port}")
val ds = GrpcUpstreams(
config.id!!,
hash,
config.role,
endpoint.host!!,
endpoint.port,
@@ -289,7 +330,12 @@ open class ConfiguredUpstreams(
}
}
private fun buildWsFactory(id: String, chain: Chain, conn: UpstreamsConfig.EthereumConnection, urls: ArrayList<URI>? = null): EthereumWsFactory? {
private fun buildWsFactory(
id: String,
chain: Chain,
conn: UpstreamsConfig.EthereumConnection,
urls: ArrayList<URI>? = null
): EthereumWsFactory? {
return conn.ws?.let { endpoint ->
val wsApi = EthereumWsFactory(
id, chain,
@@ -305,15 +351,45 @@ open class ConfiguredUpstreams(
}
}
private fun buildEthereumConnectorFactory(id: String, conn: UpstreamsConfig.EthereumConnection, chain: Chain, urls: ArrayList<URI>, forkChoice: ForkChoice, blockValidator: BlockValidator): EthereumConnectorFactory? {
private fun buildEthereumConnectorFactory(
id: String,
conn: UpstreamsConfig.EthereumConnection,
chain: Chain,
urls: ArrayList<URI>,
forkChoice: ForkChoice,
blockValidator: BlockValidator
): EthereumConnectorFactory? {
val wsFactoryApi = buildWsFactory(id, chain, conn, urls)
val httpFactory = buildHttpFactory(conn, urls)
log.info("Using ${chain.chainName} upstream, at ${urls.joinToString()}")
val connectorFactory = EthereumConnectorFactory(conn.preferHttp, wsFactoryApi, httpFactory, forkChoice, blockValidator)
val connectorFactory =
EthereumConnectorFactory(conn.preferHttp, wsFactoryApi, httpFactory, forkChoice, blockValidator)
if (!connectorFactory.isValid()) {
log.warn("Upstream configuration is invalid (probably no http endpoint)")
return null
}
return connectorFactory
}
@VisibleForTesting
private fun getHash(nodeId: Int?, obj: Any): Byte =
nodeId?.toByte() ?: (obj.hashCode() % 255).let {
if (it == 0) 1 else it
}.let { nonZeroHash ->
listOf<Function<Int, Int>>(
Function { i -> i },
Function { i -> (-i) },
Function { i -> 127 - abs(i) },
Function { i -> abs(i) - 128 },
).map {
it.apply(nonZeroHash).toByte()
}.firstOrNull {
hashes[it] != true
}?.let {
hashes[it] = true
it
} ?: (Byte.MIN_VALUE..Byte.MAX_VALUE).first {
it != 0 && hashes[it.toByte()] != true
}.toByte()
}
}

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/CurrentMultistreamHolder.kt
View File

@@ -62,7 +62,7 @@ open class CurrentMultistreamHolder(
val chain = change.chain
try {
when (BlockchainType.from(chain)) {
BlockchainType.ETHEREUM -> {
BlockchainType.EVM_POW -> {
val up = change.upstream.cast(EthereumUpstream::class.java)
val current = chainMapping[chain]
val factory = Callable<Multistream> {
@@ -70,7 +70,7 @@ open class CurrentMultistreamHolder(
}
processUpdate(change, up, current, factory)
}
BlockchainType.ETHEREUM_POS -> {
BlockchainType.EVM_POS -> {
val up = change.upstream.cast(EthereumPosUpstream::class.java)
val current = chainMapping[chain]
val factory = Callable<Multistream> {
@@ -145,9 +145,9 @@ open class CurrentMultistreamHolder(
fun setupDefaultMethods(chain: Chain): CallMethods {
val created = when (BlockchainType.from(chain)) {
BlockchainType.ETHEREUM -> DefaultEthereumMethods(chain)
BlockchainType.EVM_POW -> DefaultEthereumMethods(chain)
BlockchainType.BITCOIN -> DefaultBitcoinMethods()
BlockchainType.ETHEREUM_POS -> DefaultEthereumMethods(chain)
BlockchainType.EVM_POS -> DefaultEthereumMethods(chain)
else -> throw IllegalStateException("Unsupported chain: $chain")
}
callTargets[chain] = created

8
src/main/kotlin/io/emeraldpay/dshackle/upstream/DefaultUpstream.kt
View File

@@ -26,6 +26,7 @@ import java.util.concurrent.atomic.AtomicReference
abstract class DefaultUpstream(
private val id: String,
private val hash: Byte,
defaultLag: Long,
defaultAvail: UpstreamAvailability,
private val options: UpstreamsConfig.Options,
@@ -36,12 +37,14 @@ abstract class DefaultUpstream(
constructor(
id: String,
hash: Byte,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?
) :
this(
id,
hash,
Long.MAX_VALUE,
UpstreamAvailability.UNAVAILABLE,
options,
@@ -52,12 +55,13 @@ abstract class DefaultUpstream(
constructor(
id: String,
hash: Byte,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?,
node: QuorumForLabels.QuorumItem?
) :
this(id, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, node)
this(id, hash, Long.MAX_VALUE, UpstreamAvailability.UNAVAILABLE, options, role, targets, node)
private val status = AtomicReference(Status(defaultLag, defaultAvail, statusByLag(defaultLag, defaultAvail)))
private val statusStream = Sinks.many()
@@ -139,6 +143,8 @@ abstract class DefaultUpstream(
return targets ?: throw IllegalStateException("Methods are not set")
}
override fun nodeId(): Byte = hash
private val quorumByLabel = node?.let { QuorumForLabels(it) }
?: QuorumForLabels(QuorumForLabels.QuorumItem.empty())

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

@@ -282,6 +282,8 @@ abstract class Multistream(
return false
}
override fun nodeId(): Byte = 0
fun printStatus() {
var height: Long? = null
try {

18
src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt
View File

@@ -396,4 +396,22 @@ class Selector {
return "Matcher: ${describeInternal()}"
}
}
class SameNodeMatcher(private val upstreamHash: Byte) : Matcher {
override fun matches(up: Upstream): Boolean =
up.nodeId() == upstreamHash
override fun describeInternal(): String =
"upstream node-id=${upstreamHash.toUByte()}"
override fun toString(): String {
return "Matcher: ${describeInternal()}"
}
override fun equals(other: Any?): Boolean {
if (other === this) return true
if (other !is SameNodeMatcher) return false
return other.upstreamHash == upstreamHash
}
}
}

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

@@ -40,4 +40,6 @@ interface Upstream {
fun isGrpc(): Boolean
fun <T : Upstream> cast(selfType: Class<T>): T
fun nodeId(): Byte
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinUpstream.kt
View File

@@ -31,7 +31,7 @@ abstract class BitcoinUpstream(
callMethods: CallMethods,
node: QuorumForLabels.QuorumItem,
val esploraClient: EsploraClient? = null
) : DefaultUpstream(id, options, role, callMethods, node) {
) : DefaultUpstream(id, 0.toByte(), options, role, callMethods, node) {
constructor(
id: String,

15
src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/DefaultEthereumMethods.kt
View File

@@ -70,7 +70,15 @@ class DefaultEthereumMethods(
"eth_feeHistory"
)
private val allowedMethods = anyResponseMethods + firstValueMethods + specialMethods + headVerifiedMethods
private val filterMethods = listOf(
"eth_getFilterChanges",
"eth_newFilter",
"eth_newBlockFilter",
"eth_newPendingTransactionFilter",
"eth_uninstallFilter"
)
private val allowedMethods = anyResponseMethods + firstValueMethods + specialMethods + headVerifiedMethods + filterMethods
private val hardcodedMethods = listOf(
"net_version",
@@ -88,6 +96,7 @@ class DefaultEthereumMethods(
override fun getQuorumFor(method: String): CallQuorum {
return when {
filterMethods.contains(method) -> AlwaysQuorum()
hardcodedMethods.contains(method) -> AlwaysQuorum()
firstValueMethods.contains(method) -> AlwaysQuorum()
anyResponseMethods.contains(method) -> NotLaggingQuorum(4)
@@ -125,7 +134,7 @@ class DefaultEthereumMethods(
Chain.ETHEREUM_CLASSIC == chain -> {
"\"1\""
}
Chain.MATIC == chain -> {
Chain.POLYGON == chain -> {
"\"137\""
}
Chain.TESTNET_MORDEN == chain -> {
@@ -151,7 +160,7 @@ class DefaultEthereumMethods(
Chain.ETHEREUM == chain -> {
"\"0x1\""
}
Chain.MATIC == chain -> {
Chain.POLYGON == chain -> {
"\"0x89\""
}
Chain.TESTNET_ROPSTEN == chain -> {

16
src/main/kotlin/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelector.kt
View File

@@ -57,10 +57,26 @@ class EthereumCallSelector(
return blockTagSelector(params, 1, head)
} else if (method == "eth_getStorageAt") {
return blockTagSelector(params, 2, head)
} else if (method == "eth_getFilterChanges" || method == "eth_uninstallFilter") {
return sameUpstreamMatcher(params)
}
return Mono.empty()
}
private fun sameUpstreamMatcher(params: String): Mono<Selector.Matcher> {
val list = objectMapper.readerFor(Any::class.java).readValues<Any>(params).readAll()
if (list.isEmpty()) {
return Mono.empty()
}
val filterId = list[0].toString()
if (filterId.length < 4) {
return Mono.just(Selector.SameNodeMatcher(0.toByte()))
}
val hashHex = filterId.substring(filterId.length - 2)
val nodeId = hashHex.toInt(16)
return Mono.just(Selector.SameNodeMatcher(nodeId.toByte()))
}
private fun blockTagSelector(params: String, pos: Int, head: Head): Mono<Selector.Matcher> {
val list = objectMapper.readerFor(Any::class.java).readValues<Any>(params).readAll()
if (list.size < pos + 1) {

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

@@ -111,7 +111,7 @@ class EthereumBlockValidator : BlockValidator {
)
}
val timestampValid = it.timestamp > cur.timestamp
val timestampValid = it.timestamp >= cur.timestamp
if (!timestampValid) {
log.warn(
"Block timestamp {} not valid for {}. Must be greater than {}",

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

@@ -103,7 +103,7 @@ open class EthereumMultistream(
upstreams.filter {
matcher.matches(it)
}.takeIf { ups ->
ups.all { it.isGrpc() }
ups.isNotEmpty() && ups.all { it.isGrpc() }
}?.map {
it as GrpcUpstream
}?.map {

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

@@ -36,13 +36,14 @@ import reactor.core.Disposable
open class EthereumRpcUpstream(
id: String,
hash: Byte,
val chain: Chain,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?,
private val node: QuorumForLabels.QuorumItem?,
connectorFactory: ConnectorFactory
) : EthereumUpstream(id, options, role, targets, node), Lifecycle, Upstream, CachesEnabled {
) : EthereumUpstream(id, hash, options, role, targets, node), Lifecycle, Upstream, CachesEnabled {
private val log = LoggerFactory.getLogger(EthereumRpcUpstream::class.java)
private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions())
private val connector: EthereumConnector = connectorFactory.create(this, validator, chain)

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

@@ -24,11 +24,12 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
abstract class EthereumUpstream(
id: String,
hash: Byte,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?,
private val node: QuorumForLabels.QuorumItem?
) : DefaultUpstream(id, options, role, targets, node) {
) : DefaultUpstream(id, hash, options, role, targets, node) {
private val capabilities = if (options.providesBalance != false) {
setOf(Capability.RPC, Capability.BALANCE)

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

@@ -240,6 +240,7 @@ open class WsConnection(
.aggregateFrames(msgSizeLimit)
.receiveFrames()
.map { ByteBufInputStream(it.content()).readAllBytes() }
.filter { it.isNotEmpty() }
.flatMap {
try {
val msg = parser.parse(it)

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

@@ -98,7 +98,7 @@ open class EthereumPosMultiStream(
upstreams.filter {
matcher.matches(it)
}.takeIf { ups ->
ups.all { it.isGrpc() }
ups.isNotEmpty() && ups.all { it.isGrpc() }
}?.map {
it as GrpcUpstream
}?.map {

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosRpcUpstream.kt
View File

@@ -36,13 +36,14 @@ import reactor.core.Disposable
open class EthereumPosRpcUpstream(
id: String,
hash: Byte,
val chain: Chain,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?,
private val node: QuorumForLabels.QuorumItem?,
connectorFactory: ConnectorFactory
) : EthereumPosUpstream(id, options, role, targets, node), Lifecycle, Upstream, CachesEnabled {
) : EthereumPosUpstream(id, hash, options, role, targets, node), Lifecycle, Upstream, CachesEnabled {
private val log = LoggerFactory.getLogger(EthereumPosRpcUpstream::class.java)
private val validator: EthereumUpstreamValidator = EthereumUpstreamValidator(this, getOptions())
private val connector: EthereumConnector = connectorFactory.create(this, validator, chain)

3
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum_pos/EthereumPosUpstream.kt
View File

@@ -24,11 +24,12 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
abstract class EthereumPosUpstream(
id: String,
hash: Byte,
options: UpstreamsConfig.Options,
role: UpstreamsConfig.UpstreamRole,
targets: CallMethods?,
private val node: QuorumForLabels.QuorumItem?
) : DefaultUpstream(id, options, role, targets, node) {
) : DefaultUpstream(id, hash, options, role, targets, node) {
private val capabilities = if (options.providesBalance != false) {
setOf(Capability.RPC, Capability.BALANCE)

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

@@ -51,6 +51,7 @@ import java.util.function.Function
open class EthereumGrpcUpstream(
private val parentId: String,
hash: Byte,
role: UpstreamsConfig.UpstreamRole,
private val chain: Chain,
private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
@@ -58,6 +59,7 @@ open class EthereumGrpcUpstream(
overrideLabels: UpstreamsConfig.Labels?
) : EthereumUpstream(
"${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
hash,
UpstreamsConfig.Options.getDefaults(),
role,
null,

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

@@ -51,6 +51,7 @@ import java.util.function.Function
open class EthereumPosGrpcUpstream(
private val parentId: String,
hash: Byte,
role: UpstreamsConfig.UpstreamRole,
private val chain: Chain,
private val remote: ReactorBlockchainGrpc.ReactorBlockchainStub,
@@ -59,6 +60,7 @@ open class EthereumPosGrpcUpstream(
overrideLabels: UpstreamsConfig.Labels?
) : EthereumPosUpstream(
"${parentId}_${chain.chainCode.lowercase(Locale.getDefault())}",
hash,
UpstreamsConfig.Options.getDefaults(),
role,
null, null

9
src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcUpstreams.kt
View File

@@ -52,6 +52,7 @@ import kotlin.concurrent.withLock
class GrpcUpstreams(
private val id: String,
private val hash: Byte,
private val role: UpstreamsConfig.UpstreamRole,
private val host: String,
private val port: Int,
@@ -190,11 +191,11 @@ class GrpcUpstreams(
)
val blockchainType = BlockchainType.from(chain)
if (blockchainType == BlockchainType.ETHEREUM) {
if (blockchainType == BlockchainType.EVM_POW) {
return getOrCreateEthereum(chain, metrics)
} else if (blockchainType == BlockchainType.BITCOIN) {
return getOrCreateBitcoin(chain, metrics)
} else if (blockchainType == BlockchainType.ETHEREUM_POS) {
} else if (blockchainType == BlockchainType.EVM_POS) {
return getOrCreateEthereumPos(chain, metrics)
} else {
throw IllegalArgumentException("Unsupported blockchain: $chain")
@@ -206,7 +207,7 @@ class GrpcUpstreams(
val current = known[chain]
return if (current == null) {
val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
val created = EthereumGrpcUpstream(id, role, chain, client!!, rpcClient, labels)
val created = EthereumGrpcUpstream(id, hash, role, chain, client!!, rpcClient, labels)
created.timeout = this.timeout
known[chain] = created
created.start()
@@ -222,7 +223,7 @@ class GrpcUpstreams(
val current = known[chain]
return if (current == null) {
val rpcClient = JsonRpcGrpcClient(client!!, chain, metrics)
val created = EthereumPosGrpcUpstream(id, role, chain, client!!, rpcClient, nodeRating, labels)
val created = EthereumPosGrpcUpstream(id, hash, role, chain, client!!, rpcClient, nodeRating, labels)
created.timeout = this.timeout
known[chain] = created
created.start()

14
src/test/groovy/io/emeraldpay/dshackle/config/UpstreamsConfigReaderSpec.groovy
View File

@@ -401,4 +401,18 @@ class UpstreamsConfigReaderSpec extends Specification {
act.upstreams.get(0).role == UpstreamsConfig.UpstreamRole.PRIMARY
act.upstreams.get(1).role == UpstreamsConfig.UpstreamRole.PRIMARY
}
def "Parse node id"() {
setup:
def config = this.class.getClassLoader().getResourceAsStream("upstreams-node-id.yaml")
when:
def act = reader.read(config)
then:
act != null
act.upstreams.size() == 2
act.upstreams[0].nodeId == 1
act.upstreams[0].id == "has_node_id"
act.upstreams[1].nodeId == null
act.upstreams[1].id == "has_no_node_id"
}
}

153
src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy
View File

@@ -118,7 +118,7 @@ class NativeCallSpec extends Specification {
def nativeCall = nativeCall()
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, null, 1))
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, null, 1, Collections.singletonList((byte) 1)))
}
}
def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
@@ -395,6 +395,99 @@ class NativeCallSpec extends Specification {
}
}
def "Prepare call adds decorator for eth_newFilter"() {
setup:
def methods = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
["eth_newFilter"] as Set, [] as Set
)
methods.setQuorum("eth_newFilter", "always")
def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream())
multistream.customMethods = methods
multistream.customHead = Mock(Head)
def multistreamHolder = Mock(MultistreamHolder) {
_ * it.observeChains() >> Flux.empty()
}
def nativeCall = nativeCall(multistreamHolder)
def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChain(Common.ChainRef.CHAIN_ETHEREUM)
.addItems(
BlockchainOuterClass.NativeCallItem.newBuilder()
.setId(1)
.setMethod("eth_newFilter")
)
.build()
when:
def act = nativeCall.prepareCall(req, multistream)
.collectList().block(Duration.ofSeconds(1)).first()
then:
act instanceof NativeCall.ValidCallContext
act.resultDecorator instanceof NativeCall.CreateFilterDecorator
}
def "Prepare call adds decorator for eth_getFilterChanges"() {
setup:
def methods = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
["eth_getFilterChanges"] as Set, [] as Set
)
methods.setQuorum("eth_getFilterChanges", "always")
def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream())
multistream.customMethods = methods
multistream.customHead = Mock(Head)
def multistreamHolder = Mock(MultistreamHolder) {
_ * it.observeChains() >> Flux.empty()
}
def nativeCall = nativeCall(multistreamHolder)
def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChain(Common.ChainRef.CHAIN_ETHEREUM)
.addItems(
BlockchainOuterClass.NativeCallItem.newBuilder()
.setId(1)
.setMethod("eth_getFilterChanges")
)
.build()
when:
def act = nativeCall.prepareCall(req, multistream)
.collectList().block(Duration.ofSeconds(1)).first()
then:
act instanceof NativeCall.ValidCallContext
act.requestDecorator instanceof NativeCall.GetFilterUpdatesDecorator
}
def "Prepare call adds decorator for eth_uninstallFilter"() {
setup:
def methods = new ManagedCallMethods(
new DefaultEthereumMethods(Chain.ETHEREUM),
["eth_uninstallFilter"] as Set, [] as Set
)
methods.setQuorum("eth_uninstallFilter", "always")
def multistream = new MultistreamHolderMock.EthereumMultistreamMock(Chain.ETHEREUM, TestingCommons.upstream())
multistream.customMethods = methods
multistream.customHead = Mock(Head)
def multistreamHolder = Mock(MultistreamHolder) {
_ * it.observeChains() >> Flux.empty()
}
def nativeCall = nativeCall(multistreamHolder)
def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChain(Common.ChainRef.CHAIN_ETHEREUM)
.addItems(
BlockchainOuterClass.NativeCallItem.newBuilder()
.setId(1)
.setMethod("eth_uninstallFilter")
)
.build()
when:
def act = nativeCall.prepareCall(req, multistream)
.collectList().block(Duration.ofSeconds(1)).first()
then:
act instanceof NativeCall.ValidCallContext
act.requestDecorator instanceof NativeCall.GetFilterUpdatesDecorator
}
def "Parse empty params"() {
setup:
def nativeCall = nativeCall()
@@ -447,6 +540,64 @@ class NativeCallSpec extends Specification {
act.payload.method == "eth_test"
}
def "Decorate eth_getFilterUpdates params"() {
setup:
def nativeCall = nativeCall()
def ctx = new NativeCall.ValidCallContext(1, null, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
new NativeCall.RawCallDetails("eth_getFilterUpdates", '["0xabcd"]'),
new NativeCall.GetFilterUpdatesDecorator(), new NativeCall.NoneResultDecorator())
when:
def act = nativeCall.parseParams(ctx)
then:
act.id == 1
act.payload.params == ["0xab"]
act.payload.method == "eth_getFilterUpdates"
}
def "Decorate eth_newFilter result"() {
setup:
def quorum = new AlwaysQuorum()
def nativeCall = nativeCall()
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList((byte)255)))
}
}
def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
new NativeCall.ParsedCallDetails("eth_getFilterChanges", []),
new NativeCall.GetFilterUpdatesDecorator(), new NativeCall.CreateFilterDecorator())
when:
def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1))
def act = objectMapper.readValue(resp.result, Object)
then:
act == "0xabff"
resp.nonce == 10
}
def "Decorate eth_newFilter result with short nodeId"() {
setup:
def quorum = new AlwaysQuorum()
def nativeCall = nativeCall()
nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList((byte)1)))
}
}
def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
new NativeCall.ParsedCallDetails("eth_getFilterChanges", []),
new NativeCall.GetFilterUpdatesDecorator(), new NativeCall.CreateFilterDecorator())
when:
def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1))
def act = objectMapper.readValue(resp.result, Object)
then:
act == "0xab01"
resp.nonce == 10
}
@Ignore
//TODO
def "Calls cache before remote"() {

33
src/test/groovy/io/emeraldpay/dshackle/startup/ConfiguredUpstreamsSpec.groovy
View File

@@ -58,4 +58,37 @@ class ConfiguredUpstreamsSpec extends Specification {
act instanceof ManagedCallMethods
new String(act.executeHardcoded("foo_bar")) == "\"static_response\""
}
def "Calculate node-id"() {
setup:
def configurer = new ConfiguredUpstreams(Stub(CurrentMultistreamHolder), Stub(FileResolver), Stub(UpstreamsConfig)
)
expect:
configurer.getHash(node, src) == expected
where:
node | src | expected
1 | "" | 1
9 | "hohoho" | 9
null | "hohoho" | 120
}
def "Calculate node-id conflicting results"() {
setup:
def configurer = new ConfiguredUpstreams(Stub(CurrentMultistreamHolder), Stub(FileResolver), Stub(UpstreamsConfig)
)
when:
def h1 = configurer.getHash(null, "hohoho")
def h2 = configurer.getHash(null, "hohoho")
def h3 = configurer.getHash(null, "hohoho")
def h4 = configurer.getHash(null, "hohoho")
def h5 = configurer.getHash(null, "hohoho")
then:
h1 == (byte)120
h2 == (byte)-120
h3 == (byte)-9
h4 == (byte)8
h5 == (byte)-128
}
}

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

@@ -69,7 +69,7 @@ class EthereumPosRpcUpstreamMock extends EthereumPosRpcUpstream {
}
EthereumPosRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods, Map<String, String> labels) {
super(id, chain,
super(id, (byte)id.hashCode(), chain,
UpstreamsConfig.Options.getDefaults(),
UpstreamsConfig.UpstreamRole.PRIMARY,
methods,

2
src/test/groovy/io/emeraldpay/dshackle/test/EthereumRpcUpstreamMock.groovy
View File

@@ -60,7 +60,7 @@ class EthereumRpcUpstreamMock extends EthereumRpcUpstream {
}
EthereumRpcUpstreamMock(@NotNull String id, @NotNull Chain chain, @NotNull Reader<JsonRpcRequest, JsonRpcResponse> api, CallMethods methods) {
super(id, chain,
super(id, id.hashCode().byteValue(), chain,
UpstreamsConfig.Options.getDefaults(),
UpstreamsConfig.UpstreamRole.PRIMARY,
methods,

4
src/test/groovy/io/emeraldpay/dshackle/test/MultistreamHolderMock.groovy
View File

@@ -27,11 +27,9 @@ import io.emeraldpay.dshackle.upstream.calls.CallMethods
import io.emeraldpay.dshackle.upstream.calls.DefaultEthereumMethods
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.MultistreamHolder
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.EthereumReader
import io.emeraldpay.dshackle.upstream.ethereum.EthereumRpcUpstream
import io.emeraldpay.grpc.BlockchainType
import io.emeraldpay.grpc.Chain
import org.jetbrains.annotations.NotNull
@@ -48,7 +46,7 @@ class MultistreamHolderMock implements MultistreamHolder {
Multistream addUpstream(@NotNull Chain chain, @NotNull Upstream up) {
if (!upstreams.containsKey(chain)) {
if (BlockchainType.from(chain) == BlockchainType.ETHEREUM_POS) {
if (BlockchainType.from(chain) == BlockchainType.EVM_POS) {
if (up instanceof EthereumPosMultiStream) {
upstreams[chain] = up
} else if (up instanceof EthereumPosRpcUpstream) {

1
src/test/groovy/io/emeraldpay/dshackle/upstream/FilteredApisSpec.groovy
View File

@@ -51,6 +51,7 @@ class FilteredApisSpec extends Specification {
def connectorFactory = new EthereumConnectorFactory(false, null, httpFactory, new MostWorkForkChoice(), BlockValidator.@Companion.ALWAYS_VALID)
new EthereumRpcUpstream(
"test",
(byte)123,
Chain.ETHEREUM,
new UpstreamsConfig.Options(),
UpstreamsConfig.UpstreamRole.PRIMARY,

23
src/test/groovy/io/emeraldpay/dshackle/upstream/SelectorSpec.groovy
View File

@@ -349,4 +349,27 @@ class SelectorSpec extends Specification {
!act
}
def "Matches same nodeId"() {
setup:
def up = Mock(Upstream) {
nodeId() >> (byte)5
}
def matcher = new Selector.SameNodeMatcher((byte)5)
when:
def act = matcher.matches(up)
then:
act
}
def "Not matches nodeId"() {
setup:
def up = Mock(Upstream) {
nodeId() >> (byte)5
}
def matcher = new Selector.SameNodeMatcher((byte)1)
when:
def act = matcher.matches(up)
then:
!act
}
}

29
src/test/groovy/io/emeraldpay/dshackle/upstream/calls/EthereumCallSelectorSpec.groovy
View File

@@ -179,4 +179,33 @@ class EthereumCallSelectorSpec extends Specification {
then:
act == new Selector.HeightMatcher(100)
}
def "Get same matcher for getFilterChanges method"() {
setup:
def callSelector = new EthereumCallSelector(Mock(Reader))
def head = Mock(Head)
expect:
callSelector.getMatcher("eth_getFilterChanges", param, head).block()
== new Selector.SameNodeMatcher((byte)hash)
where:
param | hash
'["0xff09"]' | 9
'["0xff"]' | 255
'[""]' | 0
'["0x0"]' | 0
}
def "Get empty matcher for getFilterChanges method without params"() {
setup:
def callSelector = new EthereumCallSelector(Mock(Reader))
def head = Mock(Head)
when:
def act = callSelector.getMatcher("eth_getFilterChanges", "[]", head).block()
then:
act == null
}
}

16
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReaderSpec.groovy
View File

@@ -30,6 +30,7 @@ class EthereumDirectReaderSpec extends Specification {
String hash1 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5"
String address1 = "0xe0aadb0a012dbcdc529c4c743d3e0385a0b54d3d"
List<Byte> resolvers = Collections.singletonList((byte)1)
def "Reads block by hash"() {
setup:
@@ -53,8 +54,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1
)
Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers)
)
}
}
@@ -84,7 +84,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getBlockByHash", [hash1, false])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(null), null, 1
Global.objectMapper.writeValueAsBytes(null), null, 1, resolvers
)
)
}
@@ -119,7 +119,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getBlockByNumber", ["0x64", false])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1
Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers
)
)
}
@@ -155,7 +155,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(json), null, 1
Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers
)
)
}
@@ -186,7 +186,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getTransactionByHash", [hash1])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes(null), null, 1
Global.objectMapper.writeValueAsBytes(null), null, 1, resolvers
)
)
}
@@ -217,7 +217,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getBalance", [address1, "latest"])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes("0x100"), null, 1
Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolvers
)
)
}
@@ -249,7 +249,7 @@ class EthereumDirectReaderSpec extends Specification {
1 * create(_, _, _) >> Mock(Reader) {
1 * read(new JsonRpcRequest("eth_getBalance", [address1, "0xa8c9bb"])) >> Mono.just(
new QuorumRpcReader.Result(
Global.objectMapper.writeValueAsBytes("0x100"), null, 1
Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolvers
)
)
}

8
src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy
View File

@@ -50,6 +50,8 @@ class EthereumGrpcUpstreamSpec extends Specification {
Counter.builder("test2").register(TestingCommons.meterRegistry)
)
def hash = (byte)123
def "Subscribe to head"() {
setup:
def callData = [:]
@@ -81,7 +83,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
)
}
})
def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null)
def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null)
upstream.setLag(0)
upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
.setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
@@ -139,7 +141,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
)
}
})
def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null)
def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null)
upstream.setLag(0)
upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
.setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))
@@ -201,7 +203,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
finished.complete(true)
}
})
def upstream = new EthereumGrpcUpstream("test", UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null)
def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, chain, client, new JsonRpcGrpcClient(client, chain, metrics), null)
upstream.setLag(0)
upstream.update(BlockchainOuterClass.DescribeChain.newBuilder()
.setStatus(BlockchainOuterClass.ChainStatus.newBuilder().setQuorum(1).setAvailabilityValue(UpstreamAvailability.OK.grpcId))

34
src/test/resources/upstreams-node-id.yaml Normal file
View File

@@ -0,0 +1,34 @@
version: v1
upstreams:
- id: has_node_id
node-id: 1
chain: ethereum
connection:
ethereum:
rpc:
url: "http://localhost:8545"
- id: has_no_node_id
chain: ethereum
connection:
ethereum:
rpc:
url: "http://localhost:8545"
ws:
url: "ws://localhost:8546"
- id: conflicted_node_id
node-id: 1
chain: ethereum
connection:
prefer-http: true
ethereum:
rpc:
url: "http://localhost:9545"
ws:
url: "ws://localhost:9546"
- id: invalid_node_id
node-id: 256
chain: ethereum
connection:
grpc:
host: "localhost"
port: 2449