diff --git a/docs/reference-configuration.adoc b/docs/reference-configuration.adoc index f9193da9..6a0be167 100644 --- a/docs/reference-configuration.adoc +++ b/docs/reference-configuration.adoc @@ -51,6 +51,7 @@ proxy: host: 0.0.0.0 port: 8080 websocket: true + preserve-batch-order: false tls: enabled: true server: @@ -352,6 +353,7 @@ health: proxy: host: 0.0.0.0 port: 8080 + preserve-batch-order: false tls: enabled: true server: @@ -380,6 +382,10 @@ proxy: | `8080` | Port to bind HTT server +| `port` +| `false` +| Should proxy preserve request-response correspondence when sending batch request via http + | `websocket` | `true` | Enable WebSocket Proxy @@ -389,6 +395,12 @@ proxy: | Setup TLS configuration for the Proxy server. See <> section +| `preserve-batch-order` +| false +| If `false` Dshackle may produce _batch_ response in different order, which is correct as per JSON RPC Spec. +If set to `true` then Dshackle preserves _batch_ order based on request order. +Note that latter is ineffective and use this option only when a client cannot reference responses by their IDs. + | `routes` | a| Routing paths for Proxy. diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt index fc01f00d..7d11e2aa 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt @@ -50,6 +50,11 @@ open class ProxyConfig { */ var routes: List = ArrayList() + /** + * Should proxy preserve request-response correspondence when sending batch request via http + */ + var preserveBatchOrder: Boolean = false + class Route( /** * URL binding for the route. http://$host:$port/$id diff --git a/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt index 390453f0..10ef82a4 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt @@ -61,6 +61,9 @@ class ProxyConfigReader : YamlConfigReader(), ConfigReader { getValueAsBool(input, "websocket")?.let { config.websocketEnabled = it } + getValueAsBool(input, "preserve-batch-order")?.let { + config.preserveBatchOrder = it + } val currentRoutes = HashSet() getList(input, "routes")?.let { routes -> config.routes = routes.value.map { route -> diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt index a37b0648..a11dc01e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/BaseHandler.kt @@ -36,7 +36,12 @@ abstract class BaseHandler( private val log = LoggerFactory.getLogger(BaseHandler::class.java) } - fun execute(chain: Chain, call: ProxyCall, handler: AccessHandlerHttp.RequestHandler): Publisher { + fun execute( + chain: Chain, + call: ProxyCall, + handler: AccessHandlerHttp.RequestHandler, + preserveBatchOrder: Boolean = false + ): Publisher { // return empty response for empty request if (call.items.isEmpty()) { return if (call.type == ProxyCall.RpcType.BATCH) { @@ -46,11 +51,20 @@ abstract class BaseHandler( } } val jsons = execute(chain, call.items, handler) - .transform(writeRpcJson.toJsons(call)) + return if (call.type == ProxyCall.RpcType.SINGLE) { - jsons.next() + jsons.transform(writeRpcJson.toJsons(call)).next() } else { - jsons.transform(writeRpcJson.asArray()) + jsons + .let { + if (preserveBatchOrder) { + it.transform(reorderByRequest(call.items)) + } else { + it + } + } + .transform(writeRpcJson.toJsons(call)) + .transform(writeRpcJson.asArray()) } } @@ -86,4 +100,28 @@ abstract class BaseHandler( } } } + + /** + * Reorders responses to the original request order. + * Note that it's highly inefficient because it requires keeping all the responses in memory until last one is processes, so should be used only if + * a client is unable to reference responses by their IDs. + */ + fun reorderByRequest(items: List): java.util.function.Function, Flux> { + val order = items.map { it.id } + return java.util.function.Function { src -> + src.collectList() + .map { results -> + order.map { id -> + results.find { it.id == id } + // If Proxy is configured to preserve original order it means that a client expect responses at exact same position + // as requests even if a request completely failed for a some reason. It's very unlikely situation, but still possible + // At this case, if we found a gap in responses, we put a default response with an error + ?: NativeCall.CallResult(id, null, NativeCall.CallError(id, "No response", null)) + } + } + .flatMapMany { + Flux.fromIterable(it) + } + } + } } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/HttpHandler.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/HttpHandler.kt index b5aa5e29..44c58f65 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/HttpHandler.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/HttpHandler.kt @@ -37,6 +37,7 @@ import java.util.function.BiFunction * Responds to HTTP requests made to the Ethereum Proxy Server */ class HttpHandler( + private val config: ProxyConfig, private val readRpcJson: ReadRpcJson, writeRpcJson: WriteRpcJson, nativeCall: NativeCall, @@ -78,7 +79,7 @@ class HttpHandler( requestMetrics.get(chain, "invalid_method").errorMetric.increment() } .flatMapMany { call -> - execute(chain, call, handler) + execute(chain, call, handler, config.preserveBatchOrder) } .onErrorResume(RpcException::class.java) { err -> val id = err.details?.let { diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyCall.kt index e8efa5c2..f4da986f 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyCall.kt @@ -36,7 +36,7 @@ class ProxyCall( /** * Mapping from our internal ids to user provided JSON RPC ids. */ - val ids = HashMap() + val ids = ArrayList() /** * Content of the request diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt index 96198e58..629cdc97 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ProxyServer.kt @@ -80,7 +80,7 @@ class ProxyServer( StandardRequestMetrics() } - private val httpHandler = HttpHandler(readRpcJson, writeRpcJson, nativeCall, accessHandler, requestMetrics) + private val httpHandler = HttpHandler(config, readRpcJson, writeRpcJson, nativeCall, accessHandler, requestMetrics) private val wsHandler: WebsocketHandler? = if (config.websocketEnabled) { WebsocketHandler(readRpcJson, writeRpcJson, nativeCall, nativeSubscribe, accessHandler, requestMetrics) } else null diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt index feb785e9..dcdca838 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt @@ -165,10 +165,9 @@ open class ReadRpcJson : Function { var seq = seqStart return items .map { json -> - val id = seq++ - context.ids[id] = json.id + context.ids.add(json.id) BlockchainOuterClass.NativeCallItem.newBuilder() - .setId(id) + .setId(context.ids.size - 1) .setMethod(json.method) .setPayload(ByteString.copyFrom(objectMapper.writeValueAsBytes(json.params))) .build() diff --git a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt index c0ed6ee4..83f8c4d9 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/proxy/WriteRpcJson.kt @@ -45,7 +45,7 @@ open class WriteRpcJson { return Function { flux -> flux .flatMap { response -> - if (!call.ids.containsKey(response.id)) { + if (call.ids.size <= response.id) { log.warn("ID wasn't requested: ${response.id}") return@flatMap Flux.empty() } diff --git a/src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy index 6ac07a94..ee57a8da 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/config/ProxyConfigReaderSpec.groovy @@ -81,6 +81,7 @@ class ProxyConfigReaderSpec extends Specification { act.enabled act.host == '0.0.0.0' act.port == 8080 + act.preserveBatchOrder act.routes.size() == 2 with(act.routes[0]) { id == "ethereum" diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy index 607c1bab..df5f15fb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/BaseHandlerSpec.groovy @@ -34,7 +34,7 @@ class BaseHandlerSpec extends Specification { setup: def handler = new BaseHandlerImpl(new WriteRpcJson(), Stub(NativeCall), Stub(ProxyServer.RequestMetricsFactory)) when: - def act = Mono.from(handler.execute(Chain.ETHEREUM, new ProxyCall(ProxyCall.RpcType.SINGLE), requestHandler)) + def act = Mono.from(handler.execute(Chain.ETHEREUM, new ProxyCall(ProxyCall.RpcType.SINGLE), requestHandler, false)) .block(Duration.ofSeconds(1)) then: act == "" @@ -44,7 +44,7 @@ class BaseHandlerSpec extends Specification { setup: def handler = new BaseHandlerImpl(new WriteRpcJson(), Stub(NativeCall), Stub(ProxyServer.RequestMetricsFactory)) when: - def act = Mono.from(handler.execute(Chain.ETHEREUM, new ProxyCall(ProxyCall.RpcType.BATCH), requestHandler)) + def act = Mono.from(handler.execute(Chain.ETHEREUM, new ProxyCall(ProxyCall.RpcType.BATCH), requestHandler, false)) .block(Duration.ofSeconds(1)) then: act == "[]" @@ -64,7 +64,7 @@ class BaseHandlerSpec extends Specification { call.ids[0] = 5 def response = new NativeCall.CallResult(0, '{"foo": 1}'.bytes, null) when: - def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler)) + def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, false)) .collectList() .block(Duration.ofSeconds(1)) .join("") @@ -87,7 +87,7 @@ class BaseHandlerSpec extends Specification { call.ids[0] = 5 def response = new NativeCall.CallResult(0, '{"foo": 1}'.bytes, null) when: - def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler)) + def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, false)) .collectList() .block(Duration.ofSeconds(1)) .join("") @@ -96,6 +96,112 @@ class BaseHandlerSpec extends Specification { 1 * nativeCall.nativeCallResult(_) >> Flux.fromIterable([response]) } + def "Execute ordered batch call with 2 items"() { + setup: + def nativeCall = Mock(NativeCall) + def handler = new BaseHandlerImpl(new WriteRpcJson(), nativeCall, Stub(ProxyServer.RequestMetricsFactory)) + + def request1 = BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_test") + .setId(0) + .build() + def request2 = BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_test2") + .setId(1) + .build() + def call = new ProxyCall(ProxyCall.RpcType.BATCH) + call.items.add(request1) + call.ids[0] = 5 + + call.items.add(request2) + call.ids[1] = 6 + def response = [ + new NativeCall.CallResult(1, '{"foo": 2}'.bytes, null), + new NativeCall.CallResult(0, '{"foo": 1}'.bytes, null) + ] + when: + def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, true)) + .collectList() + .block(Duration.ofSeconds(1)) + .join("") + then: + act == '[{"jsonrpc":"2.0","id":5,"result":{"foo": 1}},{"jsonrpc":"2.0","id":6,"result":{"foo": 2}}]' + 1 * nativeCall.nativeCallResult(_) >> Flux.fromIterable(response) + } + + def "Execute ordered batch call with 2 items even if original ids are not in sequence"() { + setup: + def nativeCall = Mock(NativeCall) + def handler = new BaseHandlerImpl(new WriteRpcJson(), nativeCall, Stub(ProxyServer.RequestMetricsFactory)) + + def request1 = BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_test") + .setId(0) + .build() + def request2 = BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_test2") + .setId(1) + .build() + def call = new ProxyCall(ProxyCall.RpcType.BATCH) + call.items.add(request1) + call.ids[0] = 15 + + call.items.add(request2) + call.ids[1] = 6 + def response = [ + new NativeCall.CallResult(1, '{"foo": 2}'.bytes, null), + new NativeCall.CallResult(0, '{"foo": 1}'.bytes, null) + ] + when: + def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, true)) + .collectList() + .block(Duration.ofSeconds(1)) + .join("") + then: + act == '[{"jsonrpc":"2.0","id":15,"result":{"foo": 1}},{"jsonrpc":"2.0","id":6,"result":{"foo": 2}}]' + 1 * nativeCall.nativeCallResult(_) >> Flux.fromIterable(response) + } + + def "Adds a missing element if order is requests"() { + setup: + def nativeCall = Mock(NativeCall) + def handler = new BaseHandlerImpl(new WriteRpcJson(), nativeCall, Stub(ProxyServer.RequestMetricsFactory)) + + def request1 = BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_test") + .setId(0) + .build() + def request2 = BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_test2") + .setId(1) + .build() + def request3 = BlockchainOuterClass.NativeCallItem.newBuilder() + .setMethod("eth_test3") + .setId(2) + .build() + def call = new ProxyCall(ProxyCall.RpcType.BATCH) + call.items.add(request1) + call.ids[0] = 5 + call.items.add(request2) + call.ids[1] = 6 + call.items.add(request3) + call.ids[2] = 7 + + // note there is only 2 responses + def response = [ + new NativeCall.CallResult(1, '{"foo": 2}'.bytes, null), + new NativeCall.CallResult(2, '{"foo": 3}'.bytes, null) + ] + when: + def act = Flux.from(handler.execute(Chain.ETHEREUM, call, requestHandler, true)) + .collectList() + .block(Duration.ofSeconds(1)) + .join("") + then: + act == '[{"jsonrpc":"2.0","id":5,"error":{"code":-32002,"message":"No response"}},{"jsonrpc":"2.0","id":6,"result":{"foo": 2}},{"jsonrpc":"2.0","id":7,"result":{"foo": 3}}]' + 1 * nativeCall.nativeCallResult(_) >> Flux.fromIterable(response) + } + class BaseHandlerImpl extends BaseHandler { BaseHandlerImpl(@NotNull WriteRpcJson writeRpcJson, @NotNull NativeCall nativeCall, @NotNull ProxyServer.RequestMetricsFactory requestMetrics) { diff --git a/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy index e6725ddf..670640fb 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/proxy/HttpHandlerSpec.groovy @@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.proxy import com.google.protobuf.ByteString import io.emeraldpay.api.proto.BlockchainOuterClass import io.emeraldpay.api.proto.Common +import io.emeraldpay.dshackle.config.ProxyConfig import io.emeraldpay.dshackle.monitoring.accesslog.AccessHandlerHttp import io.emeraldpay.dshackle.rpc.NativeCall import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse @@ -56,6 +57,7 @@ class HttpHandlerSpec extends Specification { _ * it.create(_,) >> accessHandler } def handler = new HttpHandler( + new ProxyConfig(), new ReadRpcJson(), new WriteRpcJson(), nativeCall, accessHandlerFactory, Stub(ProxyServer.RequestMetricsFactory) ) @@ -84,6 +86,7 @@ class HttpHandlerSpec extends Specification { } def handler = new HttpHandler( + new ProxyConfig(), read, new WriteRpcJson(), Stub(NativeCall), Stub(AccessHandlerHttp.HandlerFactory), metrics @@ -110,6 +113,7 @@ class HttpHandlerSpec extends Specification { def handler = new HttpHandler( + new ProxyConfig(), new ReadRpcJson(), writeRpcJson, nativeCall, Stub(AccessHandlerHttp.HandlerFactory), Stub(ProxyServer.RequestMetricsFactory) ) @@ -122,7 +126,7 @@ class HttpHandlerSpec extends Specification { .build() ) when: - def act = handler.execute(Chain.ETHEREUM, call, new AccessHandlerHttp.NoOpHandler()) + def act = handler.execute(Chain.ETHEREUM, call, new AccessHandlerHttp.NoOpHandler(), false) then: 1 * nativeCall.nativeCallResult(_) >> Flux.just(new NativeCall.CallResult(1, "".bytes, null)) diff --git a/src/test/resources/dshackle-proxy-max.yaml b/src/test/resources/dshackle-proxy-max.yaml index 1b9fa1a0..ea95cbc3 100644 --- a/src/test/resources/dshackle-proxy-max.yaml +++ b/src/test/resources/dshackle-proxy-max.yaml @@ -2,6 +2,7 @@ proxy: enabled: true host: 0.0.0.0 port: 8080 + preserve-batch-order: true routes: - id: ethereum blockchain: ethereum diff --git a/testing/dshackle/dshackle-real.yaml b/testing/dshackle/dshackle-real.yaml index 4860527c..a8e289f0 100644 --- a/testing/dshackle/dshackle-real.yaml +++ b/testing/dshackle/dshackle-real.yaml @@ -22,6 +22,7 @@ cache: proxy: port: 18081 + preserve-batch-order: true tls: enabled: false routes: