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

Return upstreamId in response for errors too (#254)

Browse Source
This commit is contained in:
KirillPamPam
2023-07-20 16:12:40 +04:00
committed by GitHub
parent 268baf0d5b
commit 121dcf0ca7
20 changed files with 112 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -21,7 +21,6 @@ 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 {
@@ -29,8 +28,7 @@ open class AlwaysQuorum : CallQuorum {
private var result: ByteArray? = null
private var rpcError: JsonRpcError? = null
private var sig: ResponseSigner.Signature? = null
private var providedUpstreamId: String? = null
private val resolvers: MutableCollection<Upstream> = ConcurrentLinkedQueue()
private val resolvers = ArrayList<Upstream>()
override fun init(head: Head) {
}
@@ -47,21 +45,15 @@ open class AlwaysQuorum : CallQuorum {
return sig
}
override fun getProvidedUpstreamId(): String? {
return providedUpstreamId
}
override fun record(
response: ByteArray,
signature: ResponseSigner.Signature?,
upstream: Upstream,
providedUpstreamId: String?
upstream: Upstream
): Boolean {
result = response
resolved = true
sig = signature
resolvers.add(upstream)
this.providedUpstreamId = providedUpstreamId
return true
}
@@ -72,6 +64,7 @@ open class AlwaysQuorum : CallQuorum {
) {
this.rpcError = error.error
sig = signature
resolvers.add(upstream)
}
override fun getResult(): ByteArray? {
@@ -82,8 +75,7 @@ open class AlwaysQuorum : CallQuorum {
return rpcError
}
override fun getResolvedBy(): List<Upstream> =
resolvers.toList()
override fun getResolvedBy(): List<Upstream> = resolvers
override fun toString(): String {
return "Quorum: Accept Any"

14
src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt
View File

@@ -28,7 +28,6 @@ open class BroadcastQuorum(
private var txid: String? = null
private var calls = 0
private var sig: ResponseSigner.Signature? = null
private var providedUpstreamId: String? = null
override fun init(head: Head) {
}
@@ -49,23 +48,17 @@ open class BroadcastQuorum(
return sig
}
override fun getProvidedUpstreamId(): String? {
return providedUpstreamId
}
override fun recordValue(
response: ByteArray,
responseValue: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream,
providedUpstreamId: String?
upstream: Upstream
) {
calls++
if (txid == null && responseValue != null) {
txid = responseValue
sig = signature
result = response
this.providedUpstreamId = providedUpstreamId
}
}
@@ -73,16 +66,15 @@ open class BroadcastQuorum(
response: ByteArray?,
errorMessage: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream,
providedUpstreamId: String?
upstream: Upstream
) {
// can be "message: known transaction: TXID", "Transaction with the same hash was already imported" or "message: Nonce too low"
calls++
if (result == null) {
result = response
sig = signature
this.providedUpstreamId = providedUpstreamId
}
resolvers.add(upstream)
}
override fun toString(): String {

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

@@ -32,9 +32,9 @@ interface CallQuorum {
fun record(
response: ByteArray,
signature: ResponseSigner.Signature?,
upstream: Upstream,
providedUpstreamId: String?
upstream: Upstream
): Boolean
fun record(
error: JsonRpcException,
signature: ResponseSigner.Signature?,
@@ -42,7 +42,6 @@ interface CallQuorum {
)
fun getSignature(): ResponseSigner.Signature?
fun getProvidedUpstreamId(): String?
fun getResult(): ByteArray?
fun getError(): JsonRpcError?
fun getResolvedBy(): Collection<Upstream>

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

@@ -21,7 +21,6 @@ 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
/**
@@ -35,8 +34,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()
private var providedUpstreamId: String? = null
private val resolvers = ArrayList<Upstream>()
override fun init(head: Head) {
}
@@ -52,14 +50,12 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
override fun record(
response: ByteArray,
signature: ResponseSigner.Signature?,
upstream: Upstream,
providedUpstreamId: String?
upstream: Upstream
): Boolean {
val lagging = upstream.getLag()?.run { this > maxLag } ?: true
if (!lagging) {
result.set(response)
sig = signature
this.providedUpstreamId = providedUpstreamId
resolvers.add(upstream)
return true
}
@@ -76,16 +72,13 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
if (!lagging && result.get() == null) {
failed.set(true)
}
resolvers.add(upstream)
}
override fun getSignature(): ResponseSigner.Signature? {
return sig
}
override fun getProvidedUpstreamId(): String? {
return providedUpstreamId
}
override fun getResult(): ByteArray {
return result.get()
}
@@ -94,8 +87,8 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum {
return rpcError
}
override fun getResolvedBy(): Collection<Upstream> =
resolvers.toList()
override fun getResolvedBy(): Collection<Upstream> = resolvers
override fun toString(): String {
return "Quorum: late <= $maxLag blocks"
}

11
src/main/kotlin/io/emeraldpay/dshackle/quorum/NotNullQuorum.kt
View File

@@ -9,7 +9,6 @@ import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
class NotNullQuorum : CallQuorum {
private var sig: ResponseSigner.Signature? = null
private var providedUpstreamId: String? = null
private var result: ByteArray? = null
private var rpcError: JsonRpcError? = null
private val resolvers = ArrayList<Upstream>()
@@ -26,8 +25,7 @@ class NotNullQuorum : CallQuorum {
override fun record(
response: ByteArray,
signature: ResponseSigner.Signature?,
upstream: Upstream,
providedUpstreamId: String?
upstream: Upstream
): Boolean {
allFailed = false
val receivedNull = response.isEmpty() || Global.nullValue.contentEquals(response)
@@ -35,7 +33,6 @@ class NotNullQuorum : CallQuorum {
if (seenUpstreams.contains(upId) || !receivedNull) {
sig = signature
result = response
this.providedUpstreamId = providedUpstreamId
resolvers.add(upstream)
return true
}
@@ -50,22 +47,20 @@ class NotNullQuorum : CallQuorum {
rpcError = error.error
} else {
result = Global.nullValue
resolvers.add(upstream)
}
sig = signature
}
resolvers.add(upstream)
seenUpstreams.add(upId)
}
override fun getSignature(): ResponseSigner.Signature? = sig
override fun getProvidedUpstreamId(): String? = providedUpstreamId
override fun getResult(): ByteArray? = result
override fun getError(): JsonRpcError? = rpcError
override fun getResolvedBy(): Collection<Upstream> = resolvers.toList()
override fun getResolvedBy(): Collection<Upstream> = resolvers
override fun toString(): String {
return "Quorum: Not null"

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

@@ -33,8 +33,8 @@ import org.slf4j.LoggerFactory
import org.springframework.cloud.sleuth.Tracer
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import reactor.util.function.Tuple2
import reactor.util.function.Tuple3
import reactor.util.function.Tuple4
import reactor.util.function.Tuples
import java.util.Optional
import java.util.concurrent.atomic.AtomicInteger
@@ -99,8 +99,8 @@ class QuorumRpcReader(
}
private fun execute(key: JsonRpcRequest, retrySpec: reactor.util.retry.Retry): Function<Flux<Upstream>, Mono<CallQuorum>> {
val quorumReduce = BiFunction<CallQuorum, Tuple4<ByteArray, Optional<ResponseSigner.Signature>, Upstream, Optional<String>>, CallQuorum> { res, a ->
if (res.record(a.t1, a.t2.orElse(null), a.t3, a.t4.orElse(null))) {
val quorumReduce = BiFunction<CallQuorum, Tuple3<ByteArray, Optional<ResponseSigner.Signature>, Upstream>, CallQuorum> { res, a ->
if (res.record(a.t1, a.t2.orElse(null), a.t3)) {
log.trace("Quorum is resolved for method ${key.method}")
apiControl.resolve()
} else {
@@ -131,13 +131,13 @@ class QuorumRpcReader(
.filter { it.isResolved() } // return nothing if not resolved
.map { quorum ->
// TODO find actual quorum number
Result(quorum.getResult()!!, quorum.getSignature(), 1, quorum.getResolvedBy(), quorum.getProvidedUpstreamId())
Result(quorum.getResult()!!, quorum.getSignature(), 1, resolvedBy())
}
.switchIfEmpty(defaultResult)
}
}
private fun callApi(api: Upstream, key: JsonRpcRequest): Mono<Tuple4<ByteArray, Optional<ResponseSigner.Signature>, Upstream, Optional<String>>> {
private fun callApi(api: Upstream, key: JsonRpcRequest): Mono<Tuple3<ByteArray, Optional<ResponseSigner.Signature>, Upstream>> {
val apiReader = api.getIngressReader()
val spanParams = mapOf(
SPAN_REQUEST_API_TYPE to apiReader.javaClass.name,
@@ -152,10 +152,10 @@ class QuorumRpcReader(
}
// must catch not only the processing of a response but also errors thrown from the .read() call
.transform(withErrorResume(api, key))
.map { Tuples.of(it.t1, it.t2, api, it.t3) }
.map { Tuples.of(it.t1, it.t2, api) }
}
private fun withSignatureAndUpstream(api: Upstream, key: JsonRpcRequest, response: JsonRpcResponse): Function<Mono<ByteArray>, Mono<Tuple3<ByteArray, Optional<ResponseSigner.Signature>, Optional<String>>>> {
private fun withSignatureAndUpstream(api: Upstream, key: JsonRpcRequest, response: JsonRpcResponse): Function<Mono<ByteArray>, Mono<Tuple2<ByteArray, Optional<ResponseSigner.Signature>>>> {
return Function { src ->
src.map {
val signature = response.providedSignature
@@ -164,7 +164,7 @@ class QuorumRpcReader(
} else {
null
}
Tuples.of(it, Optional.ofNullable(signature), Optional.ofNullable(response.providedUpstreamId))
Tuples.of(it, Optional.ofNullable(signature))
}
}
}
@@ -201,8 +201,9 @@ class QuorumRpcReader(
private fun setupDefaultResult(key: JsonRpcRequest): Mono<Result> {
return Mono.just(quorum).flatMap { q ->
if (q.isFailed()) {
val err = q.getError()?.asException(JsonRpcResponse.NumberId(key.id))
?: JsonRpcException(JsonRpcResponse.NumberId(key.id), JsonRpcError(-32603, "Unhandled Upstream error"))
val resolvedBy = resolvedBy()?.getId()
val err = q.getError()?.asException(JsonRpcResponse.NumberId(key.id), resolvedBy)
?: JsonRpcException(JsonRpcResponse.NumberId(key.id), JsonRpcError(-32603, "Unhandled Upstream error"), resolvedBy)
log.warn("Quorum is failed. Method ${key.method}, message ${err.message}")
Mono.error(err)
} else {
@@ -212,13 +213,16 @@ class QuorumRpcReader(
}
}
private fun resolvedBy() =
if (quorum.getResolvedBy().isEmpty()) null else quorum.getResolvedBy().last()
private fun noResponse(method: String, q: CallQuorum): Mono<Result> {
return apiControl.upstreamsMatchesResponse()?.run {
tracer.currentSpan()?.tag(SPAN_NO_RESPONSE_MESSAGE, getFullCause())
val cause = getCause(method) ?: return Mono.empty()
if (cause.shouldReturnNull) {
Mono.just(
Result(Global.nullValue, null, 1, emptyList(), null)
Result(Global.nullValue, null, 1, null)
)
} else {
Mono.error(RpcException(1, "No response for method $method. Cause - ${cause.cause}"))
@@ -230,7 +234,6 @@ class QuorumRpcReader(
val value: ByteArray,
val signature: ResponseSigner.Signature?,
val quorum: Int,
val resolvers: Collection<Upstream>,
val providedUpstreamId: String?
val resolvedBy: Upstream?
)
}

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

@@ -23,7 +23,6 @@ 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>
@@ -31,7 +30,7 @@ abstract class ValueAwareQuorum<T>(
private val log = LoggerFactory.getLogger(ValueAwareQuorum::class.java)
private var rpcError: JsonRpcError? = null
private val resolvers: MutableCollection<Upstream> = ConcurrentLinkedQueue()
protected val resolvers = ArrayList<Upstream>()
fun extractValue(response: ByteArray, clazz: Class<T>): T? {
return Global.objectMapper.readValue(response.inputStream(), clazz)
@@ -40,17 +39,16 @@ abstract class ValueAwareQuorum<T>(
override fun record(
response: ByteArray,
signature: ResponseSigner.Signature?,
upstream: Upstream,
providedUpstreamId: String?
upstream: Upstream
): Boolean {
try {
val value = extractValue(response, clazz)
recordValue(response, value, signature, upstream, providedUpstreamId)
recordValue(response, value, signature, upstream)
resolvers.add(upstream)
} catch (e: RpcException) {
recordError(response, e.rpcMessage, signature, upstream, providedUpstreamId)
recordError(response, e.rpcMessage, signature, upstream)
} catch (e: Exception) {
recordError(response, e.message, signature, upstream, providedUpstreamId)
recordError(response, e.message, signature, upstream)
}
return isResolved()
}
@@ -61,29 +59,26 @@ abstract class ValueAwareQuorum<T>(
upstream: Upstream
) {
this.rpcError = error.error
recordError(null, error.error.message, signature, upstream, null)
recordError(null, error.error.message, signature, upstream)
}
abstract fun recordValue(
response: ByteArray,
responseValue: T?,
signature: ResponseSigner.Signature?,
upstream: Upstream,
providedUpstreamId: String?
upstream: Upstream
)
abstract fun recordError(
response: ByteArray?,
errorMessage: String?,
signature: ResponseSigner.Signature?,
upstream: Upstream,
providedUpstreamId: String?
upstream: Upstream
)
override fun getError(): JsonRpcError? {
return rpcError
}
override fun getResolvedBy(): Collection<Upstream> =
resolvers.toList()
override fun getResolvedBy(): Collection<Upstream> = resolvers
}

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

@@ -28,7 +28,6 @@ import io.emeraldpay.dshackle.commons.LOCAL_READER
import io.emeraldpay.dshackle.commons.REMOTE_QUORUM_RPC_READER
import io.emeraldpay.dshackle.commons.SPAN_ERROR
import io.emeraldpay.dshackle.commons.SPAN_REQUEST_ID
import io.emeraldpay.dshackle.commons.SPAN_RESPONSE_UPSTREAM_ID
import io.emeraldpay.dshackle.commons.SPAN_STATUS_MESSAGE
import io.emeraldpay.dshackle.config.MainConfig
import io.emeraldpay.dshackle.quorum.CallQuorum
@@ -131,9 +130,6 @@ open class NativeCall(
private fun completeSpan(callResult: CallResult, requestCount: Int) {
val span = tracer.currentSpan()
callResult.upstreamId?.let {
span?.tag(SPAN_RESPONSE_UPSTREAM_ID, it)
}
if (callResult.isError()) {
errorSpan(span, callResult.error?.message ?: "Internal error")
}
@@ -403,9 +399,7 @@ open class NativeCall(
.map {
val bytes = ctx.resultDecorator.processResult(it)
validateResult(bytes, "remote", ctx)
val upId = it.providedUpstreamId
?: if (it.resolvers.isEmpty()) ctx.upstream.getId() else it.resolvers.first().getId()
CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, upId, ctx)
CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, it.resolvedBy?.getId(), ctx)
}
.onErrorResume { t ->
Mono.just(CallResult.fail(ctx.id, ctx.nonce, t, ctx))
@@ -490,10 +484,8 @@ open class NativeCall(
}
override fun processResult(result: QuorumRpcReader.Result): ByteArray {
val bytes = result.value
if (bytes.last() == quoteCode) {
val suffix = result.resolvers
.map { it.nodeId() }
.first()
if (bytes.last() == quoteCode && result.resolvedBy != null) {
val suffix = result.resolvedBy.nodeId()
.toUByte()
.toString(16).padStart(2, padChar = '0').toByteArray()
bytes[bytes.lastIndex] = suffix.first()
@@ -598,7 +590,13 @@ open class NativeCall(
open class CallFailure(val id: Int, val reason: Throwable) : Exception("Failed to call $id: ${reason.message}")
open class CallError(val id: Int, val message: String, val upstreamError: JsonRpcError?, val data: String?) {
open class CallError(
val id: Int,
val message: String,
val upstreamError: JsonRpcError?,
val data: String?,
val upstreamId: String? = null
) {
companion object {
@@ -615,7 +613,7 @@ open class NativeCall(
}
fun from(t: Throwable): CallError {
return when (t) {
is JsonRpcException -> CallError(t.error.code, t.error.message, t.error, getDataAsSting(t.error.details))
is JsonRpcException -> CallError(t.error.code, t.error.message, t.error, getDataAsSting(t.error.details), t.upstreamId)
is RpcException -> CallError(t.code, t.rpcMessage, null, getDataAsSting(t.details))
is CallFailure -> CallError(t.id, t.reason.message ?: "Upstream Error", null, null)
else -> {
@@ -641,6 +639,16 @@ open class NativeCall(
val upstreamId: String?,
val ctx: ValidCallContext<ParsedCallDetails>?
) {
constructor(
id: Int,
nonce: Long?,
result: ByteArray?,
callError: CallError?,
signature: ResponseSigner.Signature?,
ctx: ValidCallContext<ParsedCallDetails>?
) : this(id, nonce, result, callError, signature, callError?.upstreamId, ctx)
companion object {
fun ok(id: Int, nonce: Long?, result: ByteArray, signature: ResponseSigner.Signature?, upstreamId: String?, ctx: ValidCallContext<ParsedCallDetails>?): CallResult {
return CallResult(id, nonce, result, null, signature, upstreamId, ctx)
@@ -651,7 +659,7 @@ open class NativeCall(
}
fun fail(id: Int, nonce: Long?, error: Throwable, ctx: ValidCallContext<ParsedCallDetails>?): CallResult {
return CallResult(id, nonce, null, CallError.from(error), null, null, ctx)
return CallResult(id, nonce, null, CallError.from(error), null, ctx)
}
}

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

@@ -377,6 +377,7 @@ open class WsConnectionImpl(
RpcResponseError.CODE_INTERNAL_ERROR,
"Response not received from WebSocket"
),
null,
false
)

6
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcError.kt
View File

@@ -31,6 +31,10 @@ data class JsonRpcError(val code: Int, val message: String, val details: Any?) {
}
fun asException(id: JsonRpcResponse.Id?): JsonRpcException {
return JsonRpcException(id ?: JsonRpcResponse.NumberId(-1), this, false)
return JsonRpcException(id ?: JsonRpcResponse.NumberId(-1), this, null, false)
}
fun asException(id: JsonRpcResponse.Id?, upstreamId: String?): JsonRpcException {
return JsonRpcException(id ?: JsonRpcResponse.NumberId(-1), this, upstreamId, false)
}
}

1
src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcException.kt
View File

@@ -20,6 +20,7 @@ import io.emeraldpay.etherjar.rpc.RpcException
class JsonRpcException(
val id: JsonRpcResponse.Id,
val error: JsonRpcError,
val upstreamId: String? = null,
writableStackTrace: Boolean = true
) : Exception(error.message, null, true, writableStackTrace) {

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

@@ -131,7 +131,7 @@ class JsonRpcHttpClient(
return Function { resp ->
resp.flatMap {
if (it.hasError()) {
Mono.error(JsonRpcException(it.id, it.error!!, false))
Mono.error(JsonRpcException(it.id, it.error!!, null, false))
} else {
Mono.just(it)
}