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

problem: no response from proxy if upstream returned error

Browse Source 
fix: #35
This commit is contained in:
Igor Artamonov
2020-07-20 00:30:52 -04:00
parent 1cca38a7bc
commit 8cefcbdb5e
29 changed files with 870 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt
View File

@@ -19,6 +19,7 @@ package io.emeraldpay.dshackle.proxy
import com.fasterxml.jackson.databind.ObjectMapper
import io.emeraldpay.api.proto.BlockchainOuterClass
import io.emeraldpay.dshackle.Global
import io.emeraldpay.dshackle.rpc.NativeCall
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import org.slf4j.LoggerFactory
import org.springframework.stereotype.Service
@@ -43,20 +44,29 @@ open class WriteRpcJson() {
*/
open fun toJsons(call: ProxyCall): Function<Flux<BlockchainOuterClass.NativeCallReplyItem>, Flux<String>> {
return Function { flux ->
flux.flatMap { response ->
if (!call.ids.containsKey(response.id)) {
log.warn("ID wasn't requested: ${response.id}")
return@flatMap Flux.empty<String>()
}
val json = toJson(call, response)
if (json == null) {
Flux.empty<String>()
} else {
Flux.just(json)
}
}.onErrorContinue { t, u ->
log.warn("Failed to convert to JSON", t)
}
flux
.flatMap { response ->
if (!call.ids.containsKey(response.id)) {
log.warn("ID wasn't requested: ${response.id}")
return@flatMap Flux.empty<String>()
}
val json = toJson(call, response)
if (json == null) {
Flux.empty<String>()
} else {
Flux.just(json)
}
}
.onErrorResume { t ->
if (t is NativeCall.CallFailure) {
Mono.just(toJson(call, t)!!)
} else {
Mono.empty()
}
}
.onErrorContinue { t, _ ->
log.warn("Failed to convert to JSON", t)
}
}
}
@@ -70,6 +80,12 @@ open class WriteRpcJson() {
return objectMapper.writeValueAsString(json)
}
fun toJson(call: ProxyCall, error: NativeCall.CallFailure): String? {
val id = call.ids[error.id] ?: return null;
val json = JsonRpcResponse.error(-32002, error.reason.message ?: "", JsonRpcResponse.Id.from(id))
return objectMapper.writeValueAsString(json)
}
/**
* Format response as JSON Array, for Batch requests
*/

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

@@ -24,16 +24,17 @@ import java.util.concurrent.atomic.AtomicReference
class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum {
private val result: AtomicReference<ByteArray> = AtomicReference()
private val failed = AtomicReference(false)
override fun init(head: Head) {
}
override fun isResolved(): Boolean {
return result.get() != null
return !isFailed() && result.get() != null
}
override fun isFailed(): Boolean {
return false
return failed.get()
}
override fun record(response: ByteArray, upstream: Upstream): Boolean {
@@ -46,6 +47,10 @@ class NotLaggingQuorum(val maxLag: Long = 0): CallQuorum {
}
override fun record(error: RpcException, upstream: Upstream) {
val lagging = upstream.getLag() > maxLag
if (!lagging && result.get() == null) {
failed.set(true)
}
}
override fun getResult(): ByteArray {

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

@@ -54,15 +54,40 @@ class QuorumRpcReader(
}
}
val defaultResult: Mono<Result> = Mono.just(quorum).flatMap { q ->
if (q.isFailed()) {
//TODO record and return actual error details
Mono.error<Result>(RpcException(-32000, "Upstream error"))
} else {
log.warn("Empty result for ${key.method} as ${q}")
Mono.empty<Result>()
}
}
return Flux.from(apis)
.takeUntil {
quorum.isFailed() || quorum.isResolved()
}
.flatMap { api ->
api.getApi().read(key)
.flatMap(JsonRpcResponse::requireResult)
// on error notify quorum, it may use error message or other details
.doOnError { err ->
if (err is RpcException) {
quorum.record(err, api)
}
api.getApi()
.read(key)
.flatMap { response ->
response.requireResult()
.onErrorResume { err ->
if (err is RpcException) {
// on error notify quorum, it may use error message or other details
quorum.record(err, api)
// it it's failed after that, then we don't need more calls, stop api source
if (quorum.isFailed()) {
apis.resolve()
} else {
apis.request(1)
}
} else {
log.warn("Result processing error", err)
}
Mono.empty()
}
}
.map { Tuples.of(it, api) }
}
@@ -95,6 +120,7 @@ class QuorumRpcReader(
// TODO find actual quorum number
QuorumRpcReader.Result(it.getResult()!!, 1)
}
.switchIfEmpty(defaultResult)
}

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

@@ -51,10 +51,12 @@ open class NativeCall(
return requestMono.flatMapMany(this::prepareCall)
.map(this::parseParams)
.parallel()
.flatMap(this::fetch)
.flatMap {
this.fetch(it)
.doOnError { e -> log.warn("Error during native call: ${e.message}") }
}
.sequential()
.map(this::buildResponse)
.doOnError { e -> log.warn("Error during native call: ${e.message}") }
.onErrorResume(this::processException)
}
@@ -63,17 +65,24 @@ open class NativeCall(
return it.withPayload(ParsedCallDetails(it.payload.method, params))
}
fun buildResponse(it: CallContext<ByteArray>): BlockchainOuterClass.NativeCallReplyItem {
return BlockchainOuterClass.NativeCallReplyItem.newBuilder()
.setSucceed(true)
fun buildResponse(it: CallResult): BlockchainOuterClass.NativeCallReplyItem {
val result = BlockchainOuterClass.NativeCallReplyItem.newBuilder()
.setSucceed(!it.isError())
.setId(it.id)
.setPayload(ByteString.copyFrom(it.payload))
.build()
if (it.isError()) {
it.error?.let { error ->
result.setErrorMessage(error.message)
}
} else {
result.setPayload(ByteString.copyFrom(it.result))
}
return result.build()
}
fun processException(it: Throwable?): Mono<BlockchainOuterClass.NativeCallReplyItem> {
val id: Int = if (it != null && CallFailure::class.isInstance(it)) {
(it as CallFailure).id
val id: Int = if (it != null && it is CallFailure) {
it.id
} else {
log.error("Lost context for a native call", it)
0
@@ -119,42 +128,52 @@ open class NativeCall(
}
}
fun fetch(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
fun fetch(ctx: CallContext<ParsedCallDetails>): Mono<CallResult> {
return ctx.upstream.getRoutedApi(ctx.matcher)
.flatMap { api ->
api.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params))
.flatMap(JsonRpcResponse::requireResult)
.map {
ctx.withPayload(it)
CallResult.ok(ctx.id, it)
}
}.switchIfEmpty(
Mono.just(ctx).flatMap(this::executeOnRemote)
)
.onErrorMap {
CallFailure(ctx.id, it)
.onErrorResume {
if (it is CallFailure) {
Mono.just(CallResult.fail(it.id, it.reason))
} else {
Mono.just(CallResult.fail(ctx.id, it))
}
}
}
fun executeOnRemote(ctx: CallContext<ParsedCallDetails>): Mono<CallContext<ByteArray>> {
fun executeOnRemote(ctx: CallContext<ParsedCallDetails>): Mono<CallResult> {
if (!ctx.upstream.getMethods().isAllowed(ctx.payload.method)) {
return Mono.error(RpcException(RpcResponseError.CODE_METHOD_NOT_EXIST, "Unsupported method"))
}
val reader = quorumReaderFactory.create(ctx.getApis(), ctx.callQuorum)
return reader.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params))
return reader
.read(JsonRpcRequest(ctx.payload.method, ctx.payload.params))
.map {
ctx.withPayload(it.value)
CallResult(ctx.id, it.value, null)
}
.doOnNext {
ctx.upstream.postprocessor
.onReceive(ctx.payload.method, ctx.payload.params, it.payload)
it.result?.let { value ->
ctx.upstream.postprocessor
.onReceive(ctx.payload.method, ctx.payload.params, value)
}
}
.onErrorMap {
log.error("Failed to make a call", it)
if (it is CallFailure) it
else CallFailure(ctx.id, it)
.onErrorResume { t ->
val failure = if (t is CallFailure) {
CallResult.fail(t.id, t.reason)
} else {
CallResult.fail(ctx.id, t)
}
Mono.just(failure)
}
.switchIfEmpty(
Mono.error(CallFailure(ctx.id, Exception("No response or no available upstream for ${ctx.payload.method}")) as Throwable)
Mono.just(CallResult.fail(ctx.id, 1, "No response or no available upstream for ${ctx.payload.method}"))
)
}
@@ -180,7 +199,39 @@ open class NativeCall(
}
}
open class CallFailure(val id: Int, val reason: Throwable): Exception("Failed to call $id: ${reason.message}")
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) {
companion object {
fun from(t: Throwable): CallError {
return when (t) {
is RpcException -> CallError(t.code, t.rpcMessage)
is CallFailure -> CallError(t.id, t.reason.message ?: "Upstream Error")
else -> CallError(1, t.message ?: "Upstream Error")
}
}
}
}
open class CallResult(val id: Int, val result: ByteArray?, val error: CallError?) {
companion object {
fun ok(id: Int, result: ByteArray): CallResult {
return CallResult(id, result, null)
}
fun fail(id: Int, errorCore: Int, errorMessage: String): CallResult {
return CallResult(id, null, CallError(errorCore, errorMessage))
}
fun fail(id: Int, error: Throwable): CallResult {
return CallResult(id, null, CallError.from(error))
}
}
fun isError(): Boolean {
return error != null
}
}
class RawCallDetails(val method: String, val params: String)
class ParsedCallDetails(val method: String, val params: List<Any>)

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

@@ -136,6 +136,7 @@ class JsonRpcResponse(
fun isInt(): Boolean
companion object {
@JvmStatic
fun from(id: Any): Id {
if (id is Int) {
return IntId(id)

18
src/test/groovy/io/emeraldpay/dshackle/quorum/NotLaggingQuorumSpec.groovy
View File

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.quorum
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.quorum.NotLaggingQuorum
import io.infinitape.etherjar.rpc.RpcException
import spock.lang.Specification
class NotLaggingQuorumSpec extends Specification {
@@ -33,6 +34,7 @@ class NotLaggingQuorumSpec extends Specification {
then:
1 * up.getLag() >> 0
quorum.isResolved()
!quorum.isFailed()
quorum.result == value
}
@@ -47,6 +49,7 @@ class NotLaggingQuorumSpec extends Specification {
then:
1 * up.getLag() >> 1
quorum.isResolved()
!quorum.isFailed()
quorum.result == value
}
@@ -61,5 +64,20 @@ class NotLaggingQuorumSpec extends Specification {
then:
1 * up.getLag() >> 2
!quorum.isResolved()
!quorum.isFailed()
}
def "Fails if no lag and error response received"() {
setup:
def up = Mock(Upstream)
def value = "foo".getBytes()
def quorum = new NotLaggingQuorum(1)
when:
quorum.record(new RpcException(-100, "test error"), up)
then:
1 * up.getLag() >> 1
!quorum.isResolved()
quorum.isFailed()
}
}

61
src/test/groovy/io/emeraldpay/dshackle/quorum/QuorumRpcReaderSpec.groovy
View File

@@ -23,6 +23,7 @@ import io.emeraldpay.dshackle.upstream.Selector
import io.emeraldpay.dshackle.upstream.Upstream
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
import io.infinitape.etherjar.rpc.RpcException
import reactor.core.publisher.Mono
import reactor.test.StepVerifier
import spock.lang.Specification
@@ -60,15 +61,16 @@ class QuorumRpcReaderSpec extends Specification {
def "always-quorum - retry upstream error"() {
setup:
def api = Mock(Reader) {
2 * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.error(1, "test")),
Mono.just(JsonRpcResponse.ok("1"))
]
}
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD
_ * getApi() >> Mock(Reader) {
2 * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.error(1, "test")),
Mono.just(JsonRpcResponse.ok("1"))
]
}
_ * getApi() >> api
}
def apis = new FilteredApis(
[up], Selector.empty
@@ -180,18 +182,19 @@ class QuorumRpcReaderSpec extends Specification {
.verify(Duration.ofSeconds(1))
}
def "non-empty-quorum - no result if all failed"() {
def "non-empty-quorum - error if all failed"() {
setup:
def api = Mock(Reader) {
3 * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.ok("null")),
Mono.just(JsonRpcResponse.error(1, "test")),
Mono.just(JsonRpcResponse.ok("null"))
]
}
def up = Mock(Upstream) {
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD
_ * getApi() >> Mock(Reader) {
3 * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.ok("null")),
Mono.just(JsonRpcResponse.error(1, "test")),
Mono.just(JsonRpcResponse.ok("null"))
]
}
_ * getApi() >> api
}
def apis = new FilteredApis(
[up], Selector.empty
@@ -206,7 +209,35 @@ class QuorumRpcReaderSpec extends Specification {
then:
StepVerifier.create(act)
.expectComplete()
.expectError()
.verify(Duration.ofSeconds(2))
}
def "Return error is upstream returned it"() {
setup:
def up = Mock(Upstream) {
_ * getLag() >> 0
_ * isAvailable() >> true
_ * getRole() >> UpstreamsConfig.UpstreamRole.STANDARD
_ * getApi() >> Mock(Reader) {
_ * read(new JsonRpcRequest("eth_test", [])) >>> [
Mono.just(JsonRpcResponse.error(-3010, "test")),
]
}
}
def apis = new FilteredApis(
[up], Selector.empty
)
def reader = new QuorumRpcReader(apis, new NotLaggingQuorum(1))
when:
def act = reader.read(new JsonRpcRequest("eth_test", []))
then:
StepVerifier.create(act)
.expectError()
//TODO verify
//.expectErrorMatches { t -> t instanceof RpcException && t.code == -3010}
.verify(Duration.ofSeconds(1))
}

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

@@ -63,7 +63,7 @@ class NativeCallSpec extends Specification {
when:
def act = nativeCall.fetch(ctx).block(Duration.ofSeconds(1))
then:
act.payload == "1".bytes
act.result == "1".bytes
}
def "Return error if router denied the requests"() {
@@ -85,12 +85,10 @@ class NativeCallSpec extends Specification {
def act = nativeCall.fetch(ctx) //.block(Duration.ofSeconds(1))
then:
StepVerifier.create(act)
.expectErrorMatches { t ->
t instanceof NativeCall.CallFailure &&
t.id == 15 &&
t.reason instanceof RpcException &&
t.reason.rpcMessage == "Test message"
.expectNextMatches { result ->
result.id == 15 && result.isError()
}
.expectComplete()
.verify(Duration.ofSeconds(1))
}
@@ -109,7 +107,7 @@ class NativeCallSpec extends Specification {
when:
def resp = nativeCall.executeOnRemote(call).block(Duration.ofSeconds(1))
def act = objectMapper.readValue(resp.payload, Object)
def act = objectMapper.readValue(resp.result, Object)
then:
act == "foo"
}
@@ -131,7 +129,10 @@ class NativeCallSpec extends Specification {
def resp = nativeCall.executeOnRemote(call)
then:
StepVerifier.create(resp)
.expectErrorMatches({ t -> t instanceof NativeCall.CallFailure && t.id == 1 })
.expectNextMatches { result ->
result.isError()
}
.expectComplete()
.verify(Duration.ofSeconds(1))
}
@@ -176,7 +177,7 @@ class NativeCallSpec extends Specification {
when:
def resp = nativeCall.buildResponse(
new NativeCall.CallContext<byte[]>(1561, TestingCommons.multistream(TestingCommons.api()), Selector.empty, new AlwaysQuorum(), objectMapper.writeValueAsBytes(json))
new NativeCall.CallResult(1561, objectMapper.writeValueAsBytes(json), null)
)
then:
resp.id == 1561