problem: some JSON RPC client cannot use IDs and process response by their order

solution: add option to preserve batch request-response correspondence order
rel: #148

src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfig.kt

@@ -50,6 +50,11 @@ open class ProxyConfig {
      */
     var routes: List<Route> = 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

src/main/kotlin/io/emeraldpay/dshackle/config/ProxyConfigReader.kt

@@ -61,6 +61,9 @@ class ProxyConfigReader : YamlConfigReader(), ConfigReader<ProxyConfig> {
         getValueAsBool(input, "websocket")?.let {
             config.websocketEnabled = it
         }
+        getValueAsBool(input, "preserve-batch-order")?.let {
+            config.preserveBatchOrder = it
+        }
         val currentRoutes = HashSet<String>()
         getList<MappingNode>(input, "routes")?.let { routes ->
             config.routes = routes.value.map { route ->

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<String> {
+    fun execute(
+        chain: Chain,
+        call: ProxyCall,
+        handler: AccessHandlerHttp.RequestHandler,
+        preserveBatchOrder: Boolean = false
+    ): Publisher<String> {
         // 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<BlockchainOuterClass.NativeCallItem>): java.util.function.Function<Flux<NativeCall.CallResult>, Flux<NativeCall.CallResult>> {
+        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)
+                }
+        }
+    }
 }

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 {

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<Int, Any>()
+    val ids = ArrayList<Any>()
 
     /**
      * Content of the request

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

src/main/kotlin/io/emeraldpay/dshackle/proxy/ReadRpcJson.kt

@@ -165,10 +165,9 @@ open class ReadRpcJson : Function<ByteArray, ProxyCall> {
         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()

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<String>()
                     }

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"

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) {

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))

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