From 8eac1178ec73175eb88be9e876966d9b85787006 Mon Sep 17 00:00:00 2001 From: a10zn8 Date: Mon, 23 Jan 2023 20:38:10 +0400 Subject: [PATCH] propagate grpc provided upstream id --- .../dshackle/quorum/AlwaysQuorum.kt | 19 +++++++++- .../dshackle/quorum/BroadcastQuorum.kt | 23 +++++++++++- .../emeraldpay/dshackle/quorum/CallQuorum.kt | 15 +++++++- .../dshackle/quorum/NonEmptyQuorum.kt | 22 ++++++++++- .../emeraldpay/dshackle/quorum/NonceQuorum.kt | 23 +++++++++++- .../dshackle/quorum/NotLaggingQuorum.kt | 20 +++++++++- .../dshackle/quorum/QuorumRpcReader.kt | 23 ++++++------ .../dshackle/quorum/ValueAwareQuorum.kt | 37 +++++++++++++++---- .../io/emeraldpay/dshackle/rpc/NativeCall.kt | 2 +- .../upstream/rpcclient/JsonRpcGrpcClient.kt | 2 +- .../upstream/rpcclient/JsonRpcResponse.kt | 9 ++--- .../dshackle/quorum/AlwaysQuorumSpec.groovy | 2 +- .../quorum/BroadcastQuorumSpec.groovy | 18 ++++----- .../dshackle/quorum/NonEmptyQuorumSpec.groovy | 13 ++----- .../dshackle/quorum/NonceQuorumSpec.groovy | 29 +++++++-------- .../quorum/NotLaggingQuorumSpec.groovy | 11 +++--- .../quorum/ValueAwareQuorumSpec.groovy | 9 ++++- .../dshackle/rpc/NativeCallSpec.groovy | 6 +-- .../dshackle/test/ApiReaderMock.groovy | 2 +- .../ethereum/EthereumCachingReaderSpec.groovy | 18 ++++----- .../rpcclient/JsonRpcResponseSpec.groovy | 16 ++++---- 21 files changed, 217 insertions(+), 102 deletions(-) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt index 74ba4456..d9bd712b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/AlwaysQuorum.kt @@ -29,6 +29,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 = ConcurrentLinkedQueue() override fun init(head: Head) { @@ -46,15 +47,29 @@ open class AlwaysQuorum : CallQuorum { return sig } - override fun record(response: ByteArray, signature: ResponseSigner.Signature?, upstream: Upstream): Boolean { + override fun getProvidedUpstreamId(): String? { + return providedUpstreamId + } + + override fun record( + response: ByteArray, + signature: ResponseSigner.Signature?, + upstream: Upstream, + providedUpstreamId: String? + ): Boolean { result = response resolved = true sig = signature resolvers.add(upstream) + this.providedUpstreamId = providedUpstreamId return true } - override fun record(error: JsonRpcException, signature: ResponseSigner.Signature?, upstream: Upstream) { + override fun record( + error: JsonRpcException, + signature: ResponseSigner.Signature?, + upstream: Upstream + ) { this.rpcError = error.error sig = signature } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt index b5cc5bbd..1578ae0b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/BroadcastQuorum.kt @@ -28,6 +28,7 @@ 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) { } @@ -48,21 +49,39 @@ open class BroadcastQuorum( return sig } - override fun recordValue(response: ByteArray, responseValue: String?, signature: ResponseSigner.Signature?, upstream: Upstream) { + override fun getProvidedUpstreamId(): String? { + return providedUpstreamId + } + + override fun recordValue( + response: ByteArray, + responseValue: String?, + signature: ResponseSigner.Signature?, + upstream: Upstream, + providedUpstreamId: String? + ) { calls++ if (txid == null && responseValue != null) { txid = responseValue sig = signature result = response + this.providedUpstreamId = providedUpstreamId } } - override fun recordError(response: ByteArray?, errorMessage: String?, signature: ResponseSigner.Signature?, upstream: Upstream) { + override fun recordError( + response: ByteArray?, + errorMessage: String?, + signature: ResponseSigner.Signature?, + upstream: Upstream, + providedUpstreamId: String? + ) { // 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 } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt index e89f2920..c8bb15ca 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/CallQuorum.kt @@ -29,9 +29,20 @@ interface CallQuorum { fun isResolved(): Boolean fun isFailed(): Boolean - fun record(response: ByteArray, signature: ResponseSigner.Signature?, upstream: Upstream): Boolean - fun record(error: JsonRpcException, signature: ResponseSigner.Signature?, upstream: Upstream) + fun record( + response: ByteArray, + signature: ResponseSigner.Signature?, + upstream: Upstream, + providedUpstreamId: String? + ): Boolean + fun record( + error: JsonRpcException, + signature: ResponseSigner.Signature?, + upstream: Upstream + ) fun getSignature(): ResponseSigner.Signature? + + fun getProvidedUpstreamId(): String? fun getResult(): ByteArray? fun getError(): JsonRpcError? fun getResolvedBy(): Collection diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt index 63709e8d..ffd07f2c 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonEmptyQuorum.kt @@ -27,6 +27,7 @@ open class NonEmptyQuorum( private var result: ByteArray? = null private var tries: Int = 0 private var sig: ResponseSigner.Signature? = null + private var providedUpstreamId: String? = null override fun init(head: Head) { } @@ -42,11 +43,22 @@ open class NonEmptyQuorum( return sig } - override fun recordValue(response: ByteArray, responseValue: Any?, signature: ResponseSigner.Signature?, upstream: Upstream) { + override fun getProvidedUpstreamId(): String? { + return providedUpstreamId + } + + override fun recordValue( + response: ByteArray, + responseValue: Any?, + signature: ResponseSigner.Signature?, + upstream: Upstream, + providedUpstreamId: String? + ) { tries++ if (responseValue != null) { result = response sig = signature + this.providedUpstreamId = providedUpstreamId } } @@ -54,7 +66,13 @@ open class NonEmptyQuorum( return result } - override fun recordError(response: ByteArray?, errorMessage: String?, sig: ResponseSigner.Signature?, upstream: Upstream) { + override fun recordError( + response: ByteArray?, + errorMessage: String?, + sig: ResponseSigner.Signature?, + upstream: Upstream, + providedUpstreamId: String? + ) { tries++ } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt index 97f195ce..b74c8905 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NonceQuorum.kt @@ -33,6 +33,7 @@ open class NonceQuorum( private var receivedTimes = 0 private var errors = 0 private var sig: ResponseSigner.Signature? = null + private var providedUpstreamId: String? = null override fun init(head: Head) { } @@ -50,7 +51,17 @@ open class NonceQuorum( return sig } - override fun recordValue(response: ByteArray, responseValue: String?, signature: ResponseSigner.Signature?, upstream: Upstream) { + override fun getProvidedUpstreamId(): String? { + return providedUpstreamId + } + + override fun recordValue( + response: ByteArray, + responseValue: String?, + signature: ResponseSigner.Signature?, + upstream: Upstream, + providedUpstreamId: String? + ) { val value = responseValue?.let { str -> HexQuantity.from(str).value.toLong() } @@ -60,9 +71,11 @@ open class NonceQuorum( resultValue = value result = response sig = signature + this.providedUpstreamId = providedUpstreamId } else if (result == null) { result = response sig = signature + this.providedUpstreamId = providedUpstreamId } } } @@ -71,7 +84,13 @@ open class NonceQuorum( return result } - override fun recordError(response: ByteArray?, errorMessage: String?, signature: ResponseSigner.Signature?, upstream: Upstream) { + override fun recordError( + response: ByteArray?, + errorMessage: String?, + signature: ResponseSigner.Signature?, + upstream: Upstream, + providedUpstreamId: String? + ) { errors++ } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt index 980f3c50..3ae6e239 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/NotLaggingQuorum.kt @@ -36,6 +36,7 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum { private var rpcError: JsonRpcError? = null private var sig: ResponseSigner.Signature? = null private val resolvers: MutableCollection = ConcurrentLinkedQueue() + private var providedUpstreamId: String? = null override fun init(head: Head) { } @@ -48,18 +49,28 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum { return failed.get() } - override fun record(response: ByteArray, signature: ResponseSigner.Signature?, upstream: Upstream): Boolean { + override fun record( + response: ByteArray, + signature: ResponseSigner.Signature?, + upstream: Upstream, + providedUpstreamId: String? + ): Boolean { val lagging = upstream.getLag() > maxLag if (!lagging) { result.set(response) sig = signature + this.providedUpstreamId = providedUpstreamId resolvers.add(upstream) return true } return false } - override fun record(error: JsonRpcException, signature: ResponseSigner.Signature?, upstream: Upstream) { + override fun record( + error: JsonRpcException, + signature: ResponseSigner.Signature?, + upstream: Upstream + ) { this.rpcError = error.error val lagging = upstream.getLag() > maxLag if (!lagging && result.get() == null) { @@ -70,6 +81,11 @@ class NotLaggingQuorum(val maxLag: Long = 0) : CallQuorum { override fun getSignature(): ResponseSigner.Signature? { return sig } + + override fun getProvidedUpstreamId(): String? { + return providedUpstreamId + } + override fun getResult(): ByteArray { return result.get() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt index 79f709b6..79d47a91 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/QuorumRpcReader.kt @@ -27,8 +27,8 @@ import io.emeraldpay.etherjar.rpc.RpcException import org.slf4j.LoggerFactory 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 @@ -90,8 +90,8 @@ class QuorumRpcReader( } fun execute(key: JsonRpcRequest, retrySpec: reactor.util.retry.Retry): Function, Mono> { - val quorumReduce = BiFunction, Upstream>, CallQuorum> { res, a -> - if (res.record(a.t1, a.t2.orElse(null), a.t3)) { + val quorumReduce = BiFunction, Upstream, Optional>, CallQuorum> { res, a -> + if (res.record(a.t1, a.t2.orElse(null), a.t3, a.t4.orElse(null))) { apiControl.resolve() } else { // quorum needs more responses, so ask api controller to make another @@ -119,25 +119,25 @@ 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()) + Result(quorum.getResult()!!, quorum.getSignature(), 1, quorum.getResolvedBy(), quorum.getProvidedUpstreamId()) } .switchIfEmpty(defaultResult) } } - fun callApi(api: Upstream, key: JsonRpcRequest): Mono, Upstream>> { + fun callApi(api: Upstream, key: JsonRpcRequest): Mono, Upstream, Optional>> { return api.getIngressReader() .read(key) .flatMap { response -> response.requireResult() - .transform(withSignature(api, key, response)) + .transform(withSignatureAndUpstream(api, key, response)) } // 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) } + .map { Tuples.of(it.t1, it.t2, api, it.t3) } } - fun withSignature(api: Upstream, key: JsonRpcRequest, response: JsonRpcResponse): Function, Mono>>> { + fun withSignatureAndUpstream(api: Upstream, key: JsonRpcRequest, response: JsonRpcResponse): Function, Mono, Optional>>> { return Function { src -> src.map { val signature = response.providedSignature @@ -146,7 +146,7 @@ class QuorumRpcReader( } else { null } - Tuples.of(it, Optional.ofNullable(signature)) + Tuples.of(it, Optional.ofNullable(signature), Optional.ofNullable(response.providedUpstreamId)) } } } @@ -165,7 +165,7 @@ class QuorumRpcReader( JsonRpcError(-32603, "Unhandled internal error: ${err.javaClass}: ${err.message}") ) } - quorum.record(cleanErr, null, api) + quorum.record(cleanErr, null, api,) // if it's failed after that, then we don't need more calls, stop api source if (quorum.isFailed()) { apiControl.resolve() @@ -198,6 +198,7 @@ class QuorumRpcReader( val value: ByteArray, val signature: ResponseSigner.Signature?, val quorum: Int, - val resolvers: Collection + val resolvers: Collection, + val providedUpstreamId: String? ) } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt b/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt index bb8c5d3b..65bce368 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/quorum/ValueAwareQuorum.kt @@ -37,27 +37,48 @@ abstract class ValueAwareQuorum( return Global.objectMapper.readValue(response.inputStream(), clazz) } - override fun record(response: ByteArray, signature: ResponseSigner.Signature?, upstream: Upstream): Boolean { + override fun record( + response: ByteArray, + signature: ResponseSigner.Signature?, + upstream: Upstream, + providedUpstreamId: String? + ): Boolean { try { val value = extractValue(response, clazz) - recordValue(response, value, signature, upstream) + recordValue(response, value, signature, upstream, providedUpstreamId) resolvers.add(upstream) } catch (e: RpcException) { - recordError(response, e.rpcMessage, signature, upstream) + recordError(response, e.rpcMessage, signature, upstream, providedUpstreamId) } catch (e: Exception) { - recordError(response, e.message, signature, upstream) + recordError(response, e.message, signature, upstream, providedUpstreamId) } return isResolved() } - override fun record(error: JsonRpcException, signature: ResponseSigner.Signature?, upstream: Upstream) { + override fun record( + error: JsonRpcException, + signature: ResponseSigner.Signature?, + upstream: Upstream + ) { this.rpcError = error.error - recordError(null, error.error.message, signature, upstream) + recordError(null, error.error.message, signature, upstream, null) } - abstract fun recordValue(response: ByteArray, responseValue: T?, signature: ResponseSigner.Signature?, upstream: Upstream) + abstract fun recordValue( + response: ByteArray, + responseValue: T?, + signature: ResponseSigner.Signature?, + upstream: Upstream, + providedUpstreamId: String? + ) - abstract fun recordError(response: ByteArray?, errorMessage: String?, signature: ResponseSigner.Signature?, upstream: Upstream) + abstract fun recordError( + response: ByteArray?, + errorMessage: String?, + signature: ResponseSigner.Signature?, + upstream: Upstream, + providedUpstreamId: String? + ) override fun getError(): JsonRpcError? { return rpcError diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index 7ee3c434..e6c4b9c1 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -312,7 +312,7 @@ open class NativeCall( .map { val bytes = ctx.resultDecorator.processResult(it) validateResult(bytes, "remote", ctx) - CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, it.resolvers.first().getId(), ctx) + CallResult.ok(ctx.id, ctx.nonce, bytes, it.signature, it.providedUpstreamId ?: it.resolvers.first().getId(), ctx) } .onErrorResume { t -> Mono.just(CallResult.fail(ctx.id, ctx.nonce, t, ctx)) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt index 8faf70aa..621948dd 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcGrpcClient.kt @@ -90,7 +90,7 @@ class JsonRpcGrpcClient( } else { null } - Mono.just(JsonRpcResponse(bytes, null, JsonRpcResponse.NumberId(0), signature)) + Mono.just(JsonRpcResponse(bytes, null, JsonRpcResponse.NumberId(0), signature, resp.upstreamId)) } else { metrics?.fails?.increment() Mono.error( diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt index 63f946fa..b6fdfbe5 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponse.kt @@ -29,7 +29,8 @@ class JsonRpcResponse( /** * When making a request through Dshackle protocol a remote may provide its signature with the response, which we keep here */ - val providedSignature: ResponseSigner.Signature? = null + val providedSignature: ResponseSigner.Signature? = null, + val providedUpstreamId: String? = null ) { constructor(result: ByteArray?, error: JsonRpcError?) : this(result, error, NumberId(0)) @@ -113,11 +114,7 @@ class JsonRpcResponse( } fun copyWithId(id: Id): JsonRpcResponse { - return JsonRpcResponse(result, error, id, providedSignature) - } - - fun copyWithSignature(signature: ResponseSigner.Signature): JsonRpcResponse { - return JsonRpcResponse(result, error, id, signature) + return JsonRpcResponse(result, error, id, providedSignature, providedUpstreamId) } override fun equals(other: Any?): Boolean { diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/AlwaysQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/AlwaysQuorumSpec.groovy index d34039c0..1b2b403e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/AlwaysQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/AlwaysQuorumSpec.groovy @@ -42,7 +42,7 @@ class AlwaysQuorumSpec extends Specification { def quorum = new AlwaysQuorum() def up = Stub(Upstream) when: - quorum.record("123".bytes, new ResponseSigner.Signature("sig1".bytes, "test", 100), up) + quorum.record("123".bytes, new ResponseSigner.Signature("sig1".bytes, "test", 100), up, null) then: quorum.isResolved() quorum.getResult() == "123".bytes diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/BroadcastQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/BroadcastQuorumSpec.groovy index c6a8305b..9cc070f3 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/BroadcastQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/BroadcastQuorumSpec.groovy @@ -43,21 +43,21 @@ class BroadcastQuorumSpec extends Specification { !q.isResolved() when: - q.record('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null, upstream1) + q.record('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null, upstream1, null) then: !q.isResolved() - 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _, _) + 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _, _, _) when: - q.record('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null, upstream2) + q.record('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null, upstream2, null) then: !q.isResolved() - 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _, _) + 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _, _, _) when: q.record(new JsonRpcException(1, "Nonce too low"), null, upstream3) then: - 1 * q.recordError(_, _, _, _) + 1 * q.recordError(_, _, _, _, _) q.isResolved() objectMapper.readValue(q.result, Object) == "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c" } @@ -78,18 +78,18 @@ class BroadcastQuorumSpec extends Specification { q.record(new JsonRpcException(1, "Internal error"), null, upstream1) then: !q.isResolved() - 1 * q.recordError(_, _, _, _) + 1 * q.recordError(_, _, _, _, _) when: - q.record('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null, upstream2) + q.record('"0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c"'.bytes, null, upstream2, null) then: !q.isResolved() - 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _, _) + 1 * q.recordValue(_, "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c", _, _, _) when: q.record(new JsonRpcException(1, "Nonce too low"), null, upstream3) then: - 1 * q.recordError(_, _, _, _) + 1 * q.recordError(_, _, _, _, _) q.isResolved() objectMapper.readValue(q.result, Object) == "0xeaa972c0d8d1ecd3e34fbbef6d34e06670e745c788bdba31c4234a1762f0378c" } diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy index d35316e1..f93ad8de 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/NonEmptyQuorumSpec.groovy @@ -61,8 +61,6 @@ class NonEmptyQuorumSpec extends Specification { setup: def q = Spy(new NonEmptyQuorum(3)) def upstream1 = Stub(Upstream) - def upstream2 = Stub(Upstream) - def upstream3 = Stub(Upstream) when: q.init(Stub(Head)) @@ -71,7 +69,7 @@ class NonEmptyQuorumSpec extends Specification { !q.isFailed() when: - q.record('"0x11"'.bytes, null, upstream1) + q.record('"0x11"'.bytes, null, upstream1, null) then: q.isResolved() !q.isFailed() @@ -82,7 +80,6 @@ class NonEmptyQuorumSpec extends Specification { def q = Spy(new NonEmptyQuorum(3)) def upstream1 = Stub(Upstream) def upstream2 = Stub(Upstream) - def upstream3 = Stub(Upstream) when: q.init(Stub(Head)) @@ -98,7 +95,7 @@ class NonEmptyQuorumSpec extends Specification { q.signature == null when: - q.record('"0x11"'.bytes, null, upstream2) + q.record('"0x11"'.bytes, null, upstream2, null) then: q.isResolved() !q.isFailed() @@ -108,8 +105,6 @@ class NonEmptyQuorumSpec extends Specification { setup: def q = Spy(new NonEmptyQuorum(3)) def upstream1 = Stub(Upstream) - def upstream2 = Stub(Upstream) - def upstream3 = Stub(Upstream) when: q.init(Stub(Head)) @@ -118,14 +113,14 @@ class NonEmptyQuorumSpec extends Specification { !q.isFailed() when: - q.record('null'.bytes, null, upstream2) + q.record('null'.bytes, null, upstream1, null) then: !q.isFailed() !q.isResolved() when: - q.record('"0x11"'.bytes, null, upstream2) + q.record('"0x11"'.bytes, null, upstream1, null) then: q.isResolved() !q.isFailed() diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/NonceQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/NonceQuorumSpec.groovy index e18a9572..1962ece0 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/NonceQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/NonceQuorumSpec.groovy @@ -18,12 +18,9 @@ package io.emeraldpay.dshackle.quorum import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global -import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.Upstream -import io.emeraldpay.dshackle.quorum.NonceQuorum import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException -import io.emeraldpay.etherjar.rpc.RpcException import spock.lang.Specification class NonceQuorumSpec extends Specification { @@ -43,21 +40,21 @@ class NonceQuorumSpec extends Specification { !q.isResolved() when: - q.record('"0x10"'.bytes, null, upstream1) + q.record('"0x10"'.bytes, null, upstream1, null) then: !q.isResolved() - 1 * q.recordValue(_, "0x10", _, _) + 1 * q.recordValue(_, "0x10", _, _, _) when: - q.record('"0x11"'.bytes, null, upstream2) + q.record('"0x11"'.bytes, null, upstream2, null) then: !q.isResolved() - 1 * q.recordValue(_, "0x11", _, _) + 1 * q.recordValue(_, "0x11", _, _, _) when: - q.record('"0x10"'.bytes, null, upstream3) + q.record('"0x10"'.bytes, null, upstream3, null) then: - 1 * q.recordValue(_, "0x10", _, _) + 1 * q.recordValue(_, "0x10", _, _, _) q.isResolved() objectMapper.readValue(q.result, Object) == "0x11" } @@ -78,24 +75,24 @@ class NonceQuorumSpec extends Specification { q.record(new JsonRpcException(1, "Internal"), null, upstream1) then: !q.isResolved() - 1 * q.recordError(_, _, _, _) + 1 * q.recordError(_, _, _, _, _) when: - q.record('"0x11"'.bytes, null, upstream2) + q.record('"0x11"'.bytes, null, upstream2, null) then: !q.isResolved() - 1 * q.recordValue(_, "0x11", _, _) + 1 * q.recordValue(_, "0x11", _, _, _) when: - q.record('"0x10"'.bytes, null, upstream3) + q.record('"0x10"'.bytes, null, upstream3, null) then: - 1 * q.recordValue(_, "0x10", _, _) + 1 * q.recordValue(_, "0x10", _, _, _) !q.isResolved() when: - q.record('"0x11"'.bytes, null, upstream1) + q.record('"0x11"'.bytes, null, upstream1, null) then: - 1 * q.recordValue(_, "0x11", _, _) + 1 * q.recordValue(_, "0x11", _, _, _) q.isResolved() objectMapper.readValue(q.result, Object) == "0x11" } diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/NotLaggingQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/NotLaggingQuorumSpec.groovy index 49399c96..551769a6 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/NotLaggingQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/NotLaggingQuorumSpec.groovy @@ -32,7 +32,7 @@ class NotLaggingQuorumSpec extends Specification { def quorum = new NotLaggingQuorum(1) when: - quorum.record(value, null, up) + quorum.record(value, null, up, null) then: 1 * up.getLag() >> 0 quorum.isResolved() @@ -40,20 +40,21 @@ class NotLaggingQuorumSpec extends Specification { quorum.result == value } - def "Keeps signature"() { + def "Keeps signature and upstream"() { setup: def up = Mock(Upstream) def value = "foo".getBytes() def quorum = new NotLaggingQuorum(1) when: - quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up) + quorum.record(value, new ResponseSigner.Signature("sig1".bytes, "test", 100), up, "test") then: 1 * up.getLag() >> 0 quorum.isResolved() !quorum.isFailed() quorum.result == value quorum.signature == new ResponseSigner.Signature("sig1".bytes, "test", 100) + quorum.providedUpstreamId == "test" } def "Resolves if ok lag"() { @@ -63,7 +64,7 @@ class NotLaggingQuorumSpec extends Specification { def quorum = new NotLaggingQuorum(1) when: - quorum.record(value, null, up) + quorum.record(value, null, up, null) then: 1 * up.getLag() >> 1 quorum.isResolved() @@ -78,7 +79,7 @@ class NotLaggingQuorumSpec extends Specification { def quorum = new NotLaggingQuorum(1) when: - quorum.record(value, null, up) + quorum.record(value, null, up, null) then: 1 * up.getLag() >> 2 !quorum.isResolved() diff --git a/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy index 9b9f9a62..c9dc3dac 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/quorum/ValueAwareQuorumSpec.groovy @@ -66,14 +66,14 @@ class ValueAwareQuorumSpec extends Specification { } @Override - void recordValue(@NotNull byte[] response, @Nullable Object responseValue, @Nullable ResponseSigner.Signature signature, @NotNull Upstream upstream) { + void recordValue(@NotNull byte[] response, @Nullable Object responseValue, @Nullable ResponseSigner.Signature signature, @NotNull Upstream upstream, @Nullable String providedUpstreamId) { } @Override - void recordError(@Nullable byte[] response, @Nullable String errorMessage, @Nullable ResponseSigner.Signature signature, @NotNull Upstream upstream) { + void recordError(@Nullable byte[] response, @Nullable String errorMessage, @Nullable ResponseSigner.Signature signature, @NotNull Upstream upstream, @Nullable String providedUpstreamId) { } @@ -101,5 +101,10 @@ class ValueAwareQuorumSpec extends Specification { boolean isFailed() { return false } + + @Override + String getProvidedUpstreamId() { + return null + } } } diff --git a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy index e2b095e7..6ccb1185 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/rpc/NativeCallSpec.groovy @@ -131,7 +131,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, Collections.singletonList(ups))) + 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, null, 1, Collections.singletonList(ups), null)) } } def call = new NativeCall.ValidCallContext(1, 10, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum, @@ -613,7 +613,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall(multistreamHolder) nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { 1 * create(_, _, _) >> Mock(Reader) { - 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList(ups))) + 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList(ups), null)) } } def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum, @@ -648,7 +648,7 @@ class NativeCallSpec extends Specification { def nativeCall = nativeCall(multistreamHolder) nativeCall.quorumReaderFactory = Mock(QuorumReaderFactory) { 1 * create(_, _, _) >> Mock(Reader) { - 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList(ups))) + 1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"0xab\"".bytes, null, 1, Collections.singletonList(ups), null)) } } def call = new NativeCall.ValidCallContext(1, 10, multistream, Selector.empty, quorum, diff --git a/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy b/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy index 9b2c073d..45d374f9 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/test/ApiReaderMock.groovy @@ -108,7 +108,7 @@ class ApiReaderMock implements Reader { } error = new JsonRpcError(-32601, "Method ${request.method} with ${request.params} is not mocked") } - return new JsonRpcResponse(result, error, JsonRpcResponse.Id.from(request.id), null) + return new JsonRpcResponse(result, error, JsonRpcResponse.Id.from(request.id), null, null) } as Callable return Mono.fromCallable(call) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy index 07d719f2..c5c9c8bf 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy @@ -55,7 +55,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, resolvers) + Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null) ) } } @@ -85,7 +85,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, resolvers + Global.objectMapper.writeValueAsBytes(null), null, 1, resolvers, null ) ) } @@ -120,7 +120,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, resolvers + Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null ) ) } @@ -156,7 +156,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, resolvers + Global.objectMapper.writeValueAsBytes(json), null, 1, resolvers, null ) ) } @@ -192,7 +192,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just( new QuorumRpcReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList() + Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList(), null ) ) } @@ -229,7 +229,7 @@ class EthereumDirectReaderSpec extends Specification { 1 * create(_, _, _) >> Mock(Reader) { 1 * read(new JsonRpcRequest("eth_getTransactionReceipt", [hash1])) >> Mono.just( new QuorumRpcReader.Result( - Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList() + Global.objectMapper.writeValueAsBytes(json), null, 1, new ArrayList(), null ) ) } @@ -257,7 +257,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, resolvers + Global.objectMapper.writeValueAsBytes(null), null, 1, resolvers, null ) ) } @@ -288,7 +288,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, resolvers + Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolvers, null ) ) } @@ -320,7 +320,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, resolvers + Global.objectMapper.writeValueAsBytes("0x100"), null, 1, resolvers, null ) ) } diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponseSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponseSpec.groovy index 957f071a..9ab30b7a 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponseSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/rpcclient/JsonRpcResponseSpec.groovy @@ -63,7 +63,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize int id and null result"() { setup: - def json = new JsonRpcResponse("null".bytes, null, new JsonRpcResponse.NumberId(1), null) + def json = new JsonRpcResponse("null".bytes, null, new JsonRpcResponse.NumberId(1), null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -72,7 +72,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize int id and string result"() { setup: - def json = new JsonRpcResponse('"Hello World"'.bytes, null, new JsonRpcResponse.NumberId(10), null) + def json = new JsonRpcResponse('"Hello World"'.bytes, null, new JsonRpcResponse.NumberId(10), null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -81,7 +81,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize int id and object result"() { setup: - def json = new JsonRpcResponse('{"foo": "Hello World", "bar": 1}'.bytes, null, new JsonRpcResponse.NumberId(101), null) + def json = new JsonRpcResponse('{"foo": "Hello World", "bar": 1}'.bytes, null, new JsonRpcResponse.NumberId(101), null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -90,7 +90,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize int id and error"() { setup: - def json = new JsonRpcResponse(null, new JsonRpcError(-32041, "Oooops"), new JsonRpcResponse.NumberId(101), null) + def json = new JsonRpcResponse(null, new JsonRpcError(-32041, "Oooops"), new JsonRpcResponse.NumberId(101), null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -99,7 +99,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize string id and null result"() { setup: - def json = new JsonRpcResponse("null".bytes, null, new JsonRpcResponse.StringId("asf01t1gg"), null) + def json = new JsonRpcResponse("null".bytes, null, new JsonRpcResponse.StringId("asf01t1gg"), null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -108,7 +108,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize string id and string result"() { setup: - def json = new JsonRpcResponse('"Hello World"'.bytes, null, new JsonRpcResponse.StringId("10"), null) + def json = new JsonRpcResponse('"Hello World"'.bytes, null, new JsonRpcResponse.StringId("10"), null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -117,7 +117,7 @@ class JsonRpcResponseSpec extends Specification { def "Serialize string id and object result"() { setup: - def json = new JsonRpcResponse('{"foo": "Hello World", "bar": 1}'.bytes, null, new JsonRpcResponse.StringId("g8gk19g"), null) + def json = new JsonRpcResponse('{"foo": "Hello World", "bar": 1}'.bytes, null, new JsonRpcResponse.StringId("g8gk19g"), null, null) when: def act = objectMapper.writeValueAsString(json) then: @@ -128,7 +128,7 @@ class JsonRpcResponseSpec extends Specification { setup: def json = new JsonRpcResponse(null, new JsonRpcError(-32041, "Oooops"), - new JsonRpcResponse.StringId("9kbo29gkaasf"), null) + new JsonRpcResponse.StringId("9kbo29gkaasf"), null, null) when: def act = objectMapper.writeValueAsString(json) then: