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

Revert "Sync upstream 7.12.2022"

Browse Source
This commit is contained in:
a10zn8
2022-12-09 14:00:45 +04:00
committed by GitHub
parent 20564e58ba
commit dca78160ba
95 changed files with 8426 additions and 446 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/kotlin/io/emeraldpay/dshackle/proxy/WebsocketHandler.kt
View File

@@ -27,6 +27,7 @@ import io.emeraldpay.dshackle.rpc.NativeSubscribe
import io.emeraldpay.etherjar.rpc.json.RequestJson
import io.emeraldpay.etherjar.rpc.json.ResponseJson
import io.netty.buffer.ByteBufInputStream
import io.netty.buffer.Unpooled
import org.reactivestreams.Publisher
import org.slf4j.LoggerFactory
import reactor.core.publisher.Flux
@@ -73,8 +74,9 @@ class WebsocketHandler(
val eventHandler = accessHandler.start(req, routeConfig.blockchain)
val responses = respond(routeConfig.blockchain, control, requests, eventHandler)
.map { Unpooled.wrappedBuffer(it.toByteArray()) }
resp.sendString(responses, Charsets.UTF_8)
resp.send(responses)
.then()
}
}

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

@@ -284,7 +284,11 @@ open class NativeCall(
Mono.just(ctx).flatMap(this::executeOnRemote)
)
.onErrorResume {
Mono.just(CallResult.fail(ctx.id, ctx.nonce, it))
if (it is CallFailure) {
Mono.just(CallResult.fail(it.id, ctx.nonce, it.reason))
} else {
Mono.just(CallResult.fail(ctx.id, ctx.nonce, it))
}
}
}
@@ -308,14 +312,21 @@ open class NativeCall(
CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, ctx.upstream.getId())
}
.onErrorResume { t ->
Mono.just(CallResult.fail(ctx.id, ctx.nonce, t))
val failure = when (t) {
is CallFailure -> CallResult.fail(t.id, ctx.nonce, t.reason)
is JsonRpcException -> CallResult.fail(ctx.id, ctx.nonce, t.error.code, t.error.message)
else -> CallResult.fail(ctx.id, ctx.nonce, t)
}
Mono.just(failure)
}
.switchIfEmpty(
Mono.fromSupplier {
counter.get().let { attempts ->
CallResult.fail(
ctx.id, ctx.nonce,
CallError(1, "No response or no available upstream for ${ctx.payload.method}", null)
ctx.id,
ctx.nonce,
1,
errorMessage(attempts, ctx.payload.method)
).also {
countFailure(attempts, ctx)
}
@@ -480,15 +491,7 @@ open class NativeCall(
is JsonRpcException -> CallError(t.id.asNumber().toInt(), t.error.message, t.error)
is RpcException -> CallError(t.code, t.rpcMessage, null)
is CallFailure -> CallError(t.id, t.reason.message ?: "Upstream Error", null)
else -> {
// May only happen if it's an unhandled exception.
// In this case try to find a meaningless details in the stack. Most important reason for doing that is to find an ID of the request
if (t.cause != null) {
from(t.cause!!)
} else {
CallError(1, t.message ?: "Upstream Error", null)
}
}
else -> CallError(1, t.message ?: "Upstream Error", null)
}
}
}
@@ -507,8 +510,8 @@ open class NativeCall(
return CallResult(id, nonce, result, null, signature, upstreamId)
}
fun fail(id: Int, nonce: Long?, error: CallError): CallResult {
return CallResult(id, nonce, null, error, null, null)
fun fail(id: Int, nonce: Long?, errorCore: Int, errorMessage: String): CallResult {
return CallResult(id, nonce, null, CallError(errorCore, errorMessage, null), null, null)
}
fun fail(id: Int, nonce: Long?, error: Throwable): CallResult {

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

@@ -98,11 +98,6 @@ abstract class DefaultUpstream(
}
fun statusByLag(lag: Long, proposed: UpstreamAvailability): UpstreamAvailability {
if (options.disableValidation == true) {
// if we specifically told that this upstream should be _always valid_ then skip
// the status calculation and trust the proposed value as is
return proposed
}
return if (proposed == UpstreamAvailability.OK) {
when {
lag > 6 -> UpstreamAvailability.SYNCING

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

@@ -405,8 +405,7 @@ open class WsConnection(
)
val response = Flux.from(rpcReceive.asFlux())
// send the request _after_ WS subscribes to the responses, otherwise the response may come before the actual subscription and be lost
.doOnRequest { sendRpc(request) }
.doOnSubscribe { sendRpc(request) }
.filter { resp -> resp.id.asNumber() == expectedId }
.take(Defaults.timeout)
.take(1)

77
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt
View File

@@ -25,8 +25,6 @@ import io.emeraldpay.dshackle.reader.Reader
import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
import io.emeraldpay.etherjar.rpc.RpcException
import io.emeraldpay.etherjar.rpc.RpcResponseError
import io.grpc.StatusRuntimeException
import org.apache.commons.lang3.time.StopWatch
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import java.util.concurrent.TimeUnit
@@ -34,7 +32,7 @@ import java.util.concurrent.TimeUnit
class JsonRpcGrpcClient(
private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val chain: Chain,
private val metrics: RpcMetrics?,
private val metrics: RpcMetrics
) {
companion object {
@@ -48,11 +46,11 @@ class JsonRpcGrpcClient(
class Executor(
private val stub: ReactorBlockchainGrpc.ReactorBlockchainStub,
private val chain: Chain,
private val metrics: RpcMetrics?
private val metrics: RpcMetrics
) : Reader<JsonRpcRequest, JsonRpcResponse> {
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
val timer = StopWatch()
var startTime: Long = 0
val req = BlockchainOuterClass.NativeCallRequest.newBuilder()
.setChainValue(chain.id)
@@ -68,58 +66,39 @@ class JsonRpcGrpcClient(
req.addItems(reqItem.build())
return Mono.just(key)
.doOnNext { timer.start() }
.flatMap {
.doOnNext {
startTime = System.nanoTime()
}.flatMap {
stub.nativeCall(req.build())
.single()
.onErrorResume(::handleError)
.flatMap(::handleResponse)
.flatMap { resp ->
if (resp.succeed) {
val bytes = resp.payload.toByteArray()
val signature = if (resp.hasSignature()) {
extractSignature(resp.signature)
} else {
null
}
Mono.just(JsonRpcResponse(bytes, null, JsonRpcResponse.NumberId(0), signature))
} else {
metrics.fails.increment()
Mono.error(
RpcException(
RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
resp.errorMessage
)
)
}
}
}
.doOnNext {
if (timer.isStarted) {
metrics?.timer?.record(timer.getTime(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS)
if (startTime > 0) {
val now = System.nanoTime()
metrics.timer.record(now - startTime, TimeUnit.NANOSECONDS)
}
}
}
fun handleResponse(resp: BlockchainOuterClass.NativeCallReplyItem): Mono<JsonRpcResponse> =
if (resp.succeed) {
val bytes = resp.payload.toByteArray()
val signature = if (resp.hasSignature()) {
extractSignature(resp.signature)
} else {
null
}
Mono.just(JsonRpcResponse(bytes, null, JsonRpcResponse.NumberId(0), signature))
} else {
metrics?.fails?.increment()
Mono.error(
RpcException(
RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
resp.errorMessage
)
)
}
fun handleError(t: Throwable): Mono<BlockchainOuterClass.NativeCallReplyItem> {
metrics?.fails?.increment()
return when (t) {
is StatusRuntimeException -> Mono.error(
RpcException(
RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
"Remote status code: ${t.status.code.name}"
)
)
else -> Mono.error(
RpcException(
RpcResponseError.CODE_UPSTREAM_CONNECTION_ERROR,
"Other connection error"
)
)
}
}
fun extractSignature(resp: NativeCallReplySignature?): ResponseSigner.Signature? {
if (resp == null || resp.signature == null || resp.signature.isEmpty || resp.upstreamId == null || resp.upstreamId.isEmpty()) {
return null

99
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcHttpClient.kt
View File

@@ -24,13 +24,10 @@ import io.netty.handler.codec.http.HttpHeaderNames
import io.netty.handler.codec.http.HttpHeaders
import io.netty.handler.ssl.SslContextBuilder
import io.netty.resolver.DefaultAddressResolverGroup
import org.apache.commons.lang3.time.StopWatch
import org.slf4j.LoggerFactory
import reactor.core.publisher.Mono
import reactor.netty.http.client.HttpClient
import reactor.netty.resources.ConnectionProvider
import reactor.util.function.Tuple2
import reactor.util.function.Tuples
import java.io.ByteArrayInputStream
import java.security.KeyStore
import java.security.cert.CertificateFactory
@@ -38,7 +35,6 @@ import java.security.cert.X509Certificate
import java.util.Base64
import java.util.concurrent.TimeUnit
import java.util.function.Consumer
import java.util.function.Function
/**
* JSON RPC client
@@ -93,95 +89,52 @@ class JsonRpcHttpClient(
this.httpClient = build
}
fun execute(request: ByteArray): Mono<Tuple2<Int, ByteArray>> {
fun execute(request: ByteArray): Mono<ByteArray> {
val response = httpClient
.post()
.uri(target)
.send(Mono.just(request).map { Unpooled.wrappedBuffer(it) })
return response.response { header, bytes ->
val statusCode = header.status().code()
bytes.aggregate().asByteArray().map {
Tuples.of(statusCode, it)
if (header.status().code() != 200) {
Mono.error(
JsonRpcException(
JsonRpcResponse.NumberId(-2),
JsonRpcError(
RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE,
"HTTP Code: ${header.status().code()}"
)
)
)
} else {
bytes.aggregate().asByteArray()
}
}.single()
}
override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
val startTime = StopWatch()
var startTime: Long = 0
return Mono.just(key)
.map(JsonRpcRequest::toJson)
.doOnNext { startTime.start() }
.doOnNext {
startTime = System.nanoTime()
}
.flatMap(this@JsonRpcHttpClient::execute)
.doOnNext {
if (startTime.isStarted) {
metrics.timer.record(startTime.nanoTime, TimeUnit.NANOSECONDS)
if (startTime > 0) {
val now = System.nanoTime()
metrics.timer.record(now - startTime, TimeUnit.NANOSECONDS)
}
}
.transform(asJsonRpcResponse(key))
.transform(convertErrors(key))
.transform(throwIfError())
}
/**
* The subscribers expect to catch an exception if the response contains JSON RPC Error. Convert it here to JsonRpcException
*/
private fun throwIfError(): Function<Mono<JsonRpcResponse>, Mono<JsonRpcResponse>> {
return Function { resp ->
resp.flatMap {
if (it.hasError()) {
Mono.error(JsonRpcException(it.id, it.error!!))
} else {
Mono.just(it)
}
}
}
}
/**
* Convert internal exceptions to standard JsonRpcException
*/
private fun convertErrors(key: JsonRpcRequest): Function<Mono<JsonRpcResponse>, Mono<JsonRpcResponse>> {
return Function { resp ->
resp.onErrorResume { t ->
.map(parser::parse)
.onErrorResume { t ->
val err = when (t) {
is RpcException -> JsonRpcException.from(t)
is JsonRpcException -> t
else -> JsonRpcException(key.id, t.message ?: t.javaClass.name)
is RpcException -> JsonRpcResponse.error(t.code, t.rpcMessage)
is JsonRpcException -> JsonRpcResponse.error(t.error, JsonRpcResponse.NumberId(1))
else -> JsonRpcResponse.error(1, t.message ?: t.javaClass.name)
}
// here we're measure the internal errors, not upstream errors
metrics.fails.increment()
Mono.error(err)
Mono.just(err)
}
}
}
/**
* Process response from the upstream and convert it to JsonRpcResponse.
* The input is a pair of (Http Status Code, Http Response Body)
*/
private fun asJsonRpcResponse(key: JsonRpcRequest): Function<Mono<Tuple2<Int, ByteArray>>, Mono<JsonRpcResponse>> {
return Function { resp ->
resp.map {
val parsed = parser.parse(it.t2)
val statusCode = it.t1
if (statusCode != 200) {
if (parsed.hasError() && parsed.error!!.code != RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE) {
// extracted the error details from the HTTP Body
parsed
} else {
// here we got a valid response with ERROR as HTTP Status Code. We assume that HTTP Status has
// a higher priority so return an error here anyway
JsonRpcResponse.error(
RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE,
"HTTP Code: $statusCode",
JsonRpcResponse.NumberId(key.id)
)
}
} else {
parsed
}
}
}
}
}

20
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseParser.kt
View File

@@ -21,7 +21,6 @@ import com.fasterxml.jackson.core.JsonParser
import com.fasterxml.jackson.core.JsonToken
import io.emeraldpay.dshackle.Global
import io.emeraldpay.etherjar.rpc.RpcResponseError
import org.apache.commons.lang3.StringUtils
import org.slf4j.LoggerFactory
import java.io.IOException
@@ -59,18 +58,15 @@ abstract class ResponseParser<T> {
} catch (e: JsonParseException) {
log.warn("Failed to parse JSON from upstream: ${e.message}")
}
return if (state.isReady) {
state
} else {
log.debug("Failed to parse `${StringUtils.abbreviateMiddle(String(json), "...", 200)}` JSON")
state.copy(
result = null,
error = JsonRpcError(
RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE,
"Invalid JSON structure: never finalized"
)
)
if (state.isReady) {
return state
}
return Preparsed(
error = JsonRpcError(
RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE,
"Invalid JSON structure: never finalized"
)
)
}
open fun process(parser: JsonParser, json: ByteArray, field: String, state: Preparsed): Preparsed {

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/ResponseWSParser.kt
View File

@@ -44,16 +44,6 @@ class ResponseWSParser : ResponseParser<ResponseWSParser.WsResponse>() {
state.error
)
}
if (state.error != null) {
return WsResponse(
// we don't have any real option because it's just an invalid value and can be anything,
// so let's suppose its Type as RPC as a most likely scenario
Type.RPC,
state.id ?: JsonRpcResponse.Id.from(0),
null,
state.error
)
}
throw IllegalStateException("State is not ready")
}