Return upstreamId from direct reader (#257)

src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt

@@ -48,7 +48,6 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumPosMultiStream
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcError
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcException
 import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
-import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.dshackle.upstream.signature.ResponseSigner
 import io.emeraldpay.etherjar.rpc.RpcException
 import io.emeraldpay.etherjar.rpc.RpcResponseError
@@ -369,13 +368,14 @@ open class NativeCall(
             .flatMap { api ->
                 SpannedReader(api, tracer, LOCAL_READER)
                     .read(JsonRpcRequest(ctx.payload.method, ctx.payload.params, ctx.nonce, ctx.forwardedSelector))
-                    .flatMap(JsonRpcResponse::requireResult)
                     .map {
-                        validateResult(it, "local", ctx)
+                        val result = it.getResult()
+                        val upstreamId = it.providedUpstreamId ?: ctx.upstream.getId()
+                        validateResult(result, "local", ctx)
                         if (ctx.nonce != null) {
-                            CallResult.ok(ctx.id, ctx.nonce, it, signer.sign(ctx.nonce, it, ctx.upstream.getId()), ctx.upstream.getId(), ctx)
+                            CallResult.ok(ctx.id, ctx.nonce, result, signer.sign(ctx.nonce, result, upstreamId), upstreamId, ctx)
                         } else {
-                            CallResult.ok(ctx.id, null, it, null, ctx.upstream.getId(), ctx)
+                            CallResult.ok(ctx.id, null, result, null, upstreamId, ctx)
                         }
                     }
             }.switchIfEmpty(

src/main/kotlin/io/emeraldpay/dshackle/rpc/TrackEthereumAddress.kt

@@ -121,6 +121,7 @@ class TrackEthereumAddress(
             .balance()
             .read(addr.address)
             .timeout(Defaults.timeout)
+            .map { it.data }
     }
 
     private fun buildResponse(address: TrackedAddress): BlockchainOuterClass.AddressBalance {

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReader.kt

@@ -19,16 +19,13 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.cache.CurrentBlockCache
-import io.emeraldpay.dshackle.cache.HeightByHashAdding
 import io.emeraldpay.dshackle.commons.CACHE_BLOCK_BY_HASH_READER
 import io.emeraldpay.dshackle.commons.CACHE_BLOCK_BY_HEIGHT_READER
-import io.emeraldpay.dshackle.commons.CACHE_HEIGHT_BY_HASH_READER
 import io.emeraldpay.dshackle.commons.CACHE_RECEIPTS_READER
 import io.emeraldpay.dshackle.commons.CACHE_TX_BY_HASH_READER
 import io.emeraldpay.dshackle.commons.DIRECT_QUORUM_RPC_READER
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
-import io.emeraldpay.dshackle.data.SourceContainer
 import io.emeraldpay.dshackle.data.TxContainer
 import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.reader.CompoundReader
@@ -39,6 +36,7 @@ import io.emeraldpay.dshackle.reader.TransformingReader
 import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.calls.CallMethods
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader.Result
 import io.emeraldpay.dshackle.upstream.ethereum.json.BlockJson
 import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionJsonSnapshot
 import io.emeraldpay.etherjar.domain.Address
@@ -47,8 +45,8 @@ import io.emeraldpay.etherjar.domain.TransactionId
 import io.emeraldpay.etherjar.domain.Wei
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import org.apache.commons.collections4.Factory
-import org.slf4j.LoggerFactory
 import org.springframework.cloud.sleuth.Tracer
+import reactor.core.publisher.Mono
 import java.util.function.Function
 
 /**
@@ -57,19 +55,16 @@ import java.util.function.Function
 open class EthereumCachingReader(
     private val up: Multistream,
     private val caches: Caches,
-    private val callMethodsFactory: Factory<CallMethods>,
+    callMethodsFactory: Factory<CallMethods>,
     private val tracer: Tracer
 ) : Lifecycle {
 
-    companion object {
-        private val log = LoggerFactory.getLogger(EthereumCachingReader::class.java)
-    }
-
     private val objectMapper: ObjectMapper = Global.objectMapper
     private val balanceCache = CurrentBlockCache<Address, Wei>()
     private val directReader = EthereumDirectReader(up, caches, balanceCache, callMethodsFactory, tracer)
 
-    val extractBlock = Function<BlockContainer, BlockJson<TransactionRefJson>> { block ->
+    private val extractBlock = Function<Result<BlockContainer>, BlockJson<TransactionRefJson>> { result ->
+        val block = result.data
         val existing = block.getParsed(BlockJson::class.java)
         if (existing != null) {
             existing.withoutTransactionDetails()
@@ -80,12 +75,9 @@ open class EthereumCachingReader(
         }
     }
 
-    val extractTx = Function<TxContainer, TransactionJsonSnapshot> { tx ->
+    private val extractTx = Function<Result<TxContainer>, TransactionJsonSnapshot> { result ->
-        tx.getParsed(TransactionJsonSnapshot::class.java) ?: objectMapper.readValue(tx.json, TransactionJsonSnapshot::class.java)
+        result.data.getParsed(TransactionJsonSnapshot::class.java)
-    }
+            ?: objectMapper.readValue(result.data.json, TransactionJsonSnapshot::class.java)
-
-    val asRaw = Function<SourceContainer, ByteArray> { tx ->
-        tx.json ?: ByteArray(0)
     }
 
     private val idToBlockHash = Function<BlockId, BlockHash> { id -> BlockHash.from(id.value) }
@@ -95,16 +87,13 @@ open class EthereumCachingReader(
     private val idToTxHash = Function<TxId, TransactionId> { id -> TransactionId.from(id.value) }
 
     private val blocksByIdAsCont = CompoundReader(
-        SpannedReader(caches.getBlocksByHash(), tracer, CACHE_BLOCK_BY_HASH_READER),
+        SpannedReader(CacheWithUpstreamIdReader(caches.getBlocksByHash()), tracer, CACHE_BLOCK_BY_HASH_READER),
         SpannedReader(RekeyingReader(idToBlockHash, directReader.blockReader), tracer, DIRECT_QUORUM_RPC_READER)
     )
 
-    private val heightByHash =
+    fun blocksByHashAsCont(): Reader<BlockHash, Result<BlockContainer>> {
-        SpannedReader(HeightByHashAdding(caches, blocksByIdAsCont), tracer, CACHE_HEIGHT_BY_HASH_READER)
-
-    fun blocksByHashAsCont(): Reader<BlockHash, BlockContainer> {
         return CompoundReader(
-            SpannedReader(RekeyingReader(blockHashToId, caches.getBlocksByHash()), tracer, CACHE_BLOCK_BY_HASH_READER),
+            SpannedReader(CacheWithUpstreamIdReader(RekeyingReader(blockHashToId, caches.getBlocksByHash())), tracer, CACHE_BLOCK_BY_HASH_READER),
             SpannedReader(directReader.blockReader, tracer, DIRECT_QUORUM_RPC_READER)
         )
     }
@@ -116,20 +105,13 @@ open class EthereumCachingReader(
         )
     }
 
-    fun blocksByIdParsed(): Reader<BlockId, BlockJson<TransactionRefJson>> {
+    open fun blocksByIdAsCont(): Reader<BlockId, Result<BlockContainer>> {
-        return TransformingReader(
-            blocksByIdAsCont(),
-            extractBlock
-        )
-    }
-
-    open fun blocksByIdAsCont(): Reader<BlockId, BlockContainer> {
         return blocksByIdAsCont
     }
 
-    open fun blocksByHeightAsCont(): Reader<Long, BlockContainer> {
+    open fun blocksByHeightAsCont(): Reader<Long, Result<BlockContainer>> {
         return CompoundReader(
-            SpannedReader(caches.getBlocksByHeight(), tracer, CACHE_BLOCK_BY_HEIGHT_READER),
+            SpannedReader(CacheWithUpstreamIdReader(caches.getBlocksByHeight()), tracer, CACHE_BLOCK_BY_HEIGHT_READER),
             SpannedReader(directReader.blockByHeightReader, tracer, DIRECT_QUORUM_RPC_READER)
         )
     }
@@ -144,42 +126,38 @@ open class EthereumCachingReader(
     open fun txByHash(): Reader<TransactionId, TransactionJsonSnapshot> {
         return TransformingReader(
             CompoundReader(
-                RekeyingReader(txHashToId, caches.getTxByHash()),
+                CacheWithUpstreamIdReader(RekeyingReader(txHashToId, caches.getTxByHash())),
                 directReader.txReader
             ),
             extractTx
         )
     }
 
-    open fun txByHashAsCont(): Reader<TxId, TxContainer> {
+    open fun txByHashAsCont(): Reader<TxId, Result<TxContainer>> {
         return CompoundReader(
-            SpannedReader(caches.getTxByHash(), tracer, CACHE_TX_BY_HASH_READER),
+            CacheWithUpstreamIdReader(SpannedReader(caches.getTxByHash(), tracer, CACHE_TX_BY_HASH_READER)),
             SpannedReader(RekeyingReader(idToTxHash, directReader.txReader), tracer, DIRECT_QUORUM_RPC_READER)
         )
     }
 
-    fun balance(): Reader<Address, Wei> {
+    fun balance(): Reader<Address, Result<Wei>> {
         // TODO include height as part of cache?
         return CompoundReader(
-            balanceCache, directReader.balanceReader
+            CacheWithUpstreamIdReader(balanceCache), directReader.balanceReader
         )
     }
 
-    fun receipts(): Reader<TxId, ByteArray> {
+    fun receipts(): Reader<TxId, Result<ByteArray>> {
         val requested = RekeyingReader(
             { txid: TxId -> TransactionId.from(txid.value) },
             directReader.receiptReader
         )
         return CompoundReader(
-            SpannedReader(caches.getReceipts(), tracer, CACHE_RECEIPTS_READER),
+            CacheWithUpstreamIdReader(SpannedReader(caches.getReceipts(), tracer, CACHE_RECEIPTS_READER)),
             SpannedReader(requested, tracer, DIRECT_QUORUM_RPC_READER)
         )
     }
 
-    fun heightByHash(): Reader<BlockId, Long> {
-        return heightByHash
-    }
-
     override fun isRunning(): Boolean {
         // TODO should be always running?
         return true // up.isRunning
@@ -194,4 +172,13 @@ open class EthereumCachingReader(
 
     override fun stop() {
     }
+
+    private class CacheWithUpstreamIdReader<K, D>(
+        private val reader: Reader<K, D>
+    ) : Reader<K, Result<D>> {
+        override fun read(key: K): Mono<Result<D>> {
+            return reader.read(key)
+                .map { Result(it, null) }
+        }
+    }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumDirectReader.kt

@@ -53,74 +53,79 @@ class EthereumDirectReader(
     private val objectMapper: ObjectMapper = Global.objectMapper
     var quorumReaderFactory: QuorumReaderFactory = QuorumReaderFactory.default()
 
-    val blockReader: Reader<BlockHash, BlockContainer>
+    val blockReader: Reader<BlockHash, Result<BlockContainer>>
-    val blockByHeightReader: Reader<Long, BlockContainer>
+    val blockByHeightReader: Reader<Long, Result<BlockContainer>>
-    val txReader: Reader<TransactionId, TxContainer>
+    val txReader: Reader<TransactionId, Result<TxContainer>>
-    val balanceReader: Reader<Address, Wei>
+    val balanceReader: Reader<Address, Result<Wei>>
-    val receiptReader: Reader<TransactionId, ByteArray>
+    val receiptReader: Reader<TransactionId, Result<ByteArray>>
 
     init {
-        blockReader = object : Reader<BlockHash, BlockContainer> {
+        blockReader = object : Reader<BlockHash, Result<BlockContainer>> {
-            override fun read(key: BlockHash): Mono<BlockContainer> {
+            override fun read(key: BlockHash): Mono<Result<BlockContainer>> {
                 val request = JsonRpcRequest("eth_getBlockByHash", listOf(key.toHex(), false))
                 return readBlock(request, key.toHex())
             }
         }
-        blockByHeightReader = object : Reader<Long, BlockContainer> {
+        blockByHeightReader = object : Reader<Long, Result<BlockContainer>> {
-            override fun read(key: Long): Mono<BlockContainer> {
+            override fun read(key: Long): Mono<Result<BlockContainer>> {
                 val heightMatcher = Selector.HeightMatcher(key)
                 val request = JsonRpcRequest("eth_getBlockByNumber", listOf(HexQuantity.from(key).toHex(), false))
                 return readBlock(request, key.toString(), heightMatcher)
             }
         }
-        txReader = object : Reader<TransactionId, TxContainer> {
+        txReader = object : Reader<TransactionId, Result<TxContainer>> {
-            override fun read(key: TransactionId): Mono<TxContainer> {
+            override fun read(key: TransactionId): Mono<Result<TxContainer>> {
                 val request = JsonRpcRequest("eth_getTransactionByHash", listOf(key.toHex()))
                 return readWithQuorum(request) // retries were removed because we use NotNullQuorum which handle errors too
                     .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Tx not read $key")))
-                    .flatMap { txbytes ->
+                    .flatMap { result ->
-                        val tx = objectMapper.readValue(txbytes, TransactionJsonSnapshot::class.java)
+                        val tx = objectMapper.readValue(result.data, TransactionJsonSnapshot::class.java)
                         if (tx == null) {
                             Mono.empty()
                         } else {
-                            Mono.just(TxContainer.from(tx, txbytes))
+                            Mono.just(
+                                Result(TxContainer.from(tx, result.data), result.upstreamId)
+                            )
                         }
                     }
                     .doOnNext { tx ->
-                        if (tx.blockId != null) {
+                        if (tx.data.blockId != null) {
-                            caches.cache(Caches.Tag.REQUESTED, tx)
+                            caches.cache(Caches.Tag.REQUESTED, tx.data)
                         }
                     }
             }
         }
-        balanceReader = object : Reader<Address, Wei> {
+        balanceReader = object : Reader<Address, Result<Wei>> {
-            override fun read(key: Address): Mono<Wei> {
+            override fun read(key: Address): Mono<Result<Wei>> {
                 val height = up.getHead().getCurrentHeight()?.let { HexQuantity.from(it).toHex() } ?: "latest"
                 val request = JsonRpcRequest("eth_getBalance", listOf(key.toHex(), height))
                 return readWithQuorum(request)
                     .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Balance not read $key")))
                     .map {
-                        val str = String(it)
+                        val str = String(it.data)
                         // it's a json string, i.e. wrapped with quotes, ex. _"0x1234"_
                         if (str.startsWith("\"") && str.endsWith("\"")) {
-                            Wei.from(str.substring(1, str.length - 1))
+                            Result(
+                                Wei.from(str.substring(1, str.length - 1)),
+                                it.upstreamId
+                            )
                         } else {
                             throw RpcException(RpcResponseError.CODE_UPSTREAM_INVALID_RESPONSE, "Not Wei value")
                         }
                     }
                     .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(200)))
                     .doOnNext { value ->
-                        balanceCache.put(key, value)
+                        balanceCache.put(key, value.data)
                     }
             }
         }
-        receiptReader = object : Reader<TransactionId, ByteArray> {
+        receiptReader = object : Reader<TransactionId, Result<ByteArray>> {
-            override fun read(key: TransactionId): Mono<ByteArray> {
+            override fun read(key: TransactionId): Mono<Result<ByteArray>> {
                 val request = JsonRpcRequest("eth_getTransactionReceipt", listOf(key.toHex()))
                 return readWithQuorum(request)
                     .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Receipt not read $key")))
-                    .flatMap { json ->
+                    .flatMap { result ->
-                        val receipt = objectMapper.readValue(json, TransactionReceiptJson::class.java)
+                        val receipt = objectMapper.readValue(result.data, TransactionReceiptJson::class.java)
                         if (receipt == null) {
                             log.debug("Empty receipt for txId $key")
                             Mono.empty()
@@ -131,11 +136,13 @@ class EthereumDirectReader(
                                     txId = TxId.from(key),
                                     blockId = BlockId.from(receipt.blockHash),
                                     height = receipt.blockNumber,
-                                    json = json,
+                                    json = result.data,
                                     parsed = receipt
                                 )
                             )
-                            Mono.just(json)
+                            Mono.just(
+                                result
+                            )
                         }
                     }
             }
@@ -147,27 +154,35 @@ class EthereumDirectReader(
         request: JsonRpcRequest,
         id: String,
         matcher: Selector.Matcher = Selector.empty
-    ): Mono<BlockContainer> {
+    ): Mono<Result<BlockContainer>> {
         return readWithQuorum(request, matcher)
             .timeout(Defaults.timeoutInternal, Mono.error(TimeoutException("Block not read $id")))
             .retryWhen(Retry.fixedDelay(3, Duration.ofMillis(200)))
-            .flatMap { blockbytes ->
+            .flatMap { result ->
-                val block = objectMapper.readValue(blockbytes, BlockJson::class.java) as BlockJson<TransactionRefJson>?
+                val block = objectMapper.readValue(result.data, BlockJson::class.java) as BlockJson<TransactionRefJson>?
                 if (block == null) {
-                    Mono.empty<BlockContainer>()
+                    Mono.empty()
                 } else {
-                    Mono.just(BlockContainer.from(block, blockbytes, "unknown"))
+                    Mono.just(
+                        Result(
+                            BlockContainer.from(block, result.data, "unknown"),
+                            result.upstreamId
+                        )
+                    )
                 }
             }
             .doOnNext { block ->
-                caches.cache(Caches.Tag.REQUESTED, block)
+                caches.cache(Caches.Tag.REQUESTED, block.data)
             }
     }
 
     /**
      * Read from an Upstream applying a Quorum specific for that request
      */
-    private fun readWithQuorum(request: JsonRpcRequest, matcher: Selector.Matcher = Selector.empty): Mono<ByteArray> {
+    private fun readWithQuorum(
+        request: JsonRpcRequest,
+        matcher: Selector.Matcher = Selector.empty
+    ): Mono<Result<ByteArray>> {
         return Mono.just(quorumReaderFactory)
             .map {
                 it.create(
@@ -185,7 +200,12 @@ class EthereumDirectReader(
             }.flatMap {
                 it.read(request)
             }.map {
-                it.value
+                Result(it.value, it.resolvedBy?.getId())
             }
     }
+
+    data class Result<T>(
+        val data: T,
+        val upstreamId: String?
+    )
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReader.kt

@@ -26,7 +26,6 @@ import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.etherjar.hex.HexQuantity
 import io.emeraldpay.etherjar.rpc.RpcException
 import io.emeraldpay.etherjar.rpc.RpcResponseError
-import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
 import reactor.kotlin.core.publisher.switchIfEmpty
 import java.math.BigInteger
@@ -45,10 +44,6 @@ class EthereumLocalReader(
     private val localEnabled: Boolean
 ) : JsonRpcReader {
 
-    companion object {
-        private val log = LoggerFactory.getLogger(EthereumLocalReader::class.java)
-    }
-
     override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
         if (methods.isHardcoded(key.method)) {
             return Mono.just(methods.executeHardcoded(key.method))
@@ -66,7 +61,7 @@ class EthereumLocalReader(
         }
         val common = commonRequests(key)
         if (common != null) {
-            return common.map { JsonRpcResponse(it, null) }
+            return common.map { JsonRpcResponse(it.first, null, it.second) }
         }
         return Mono.empty()
     }
@@ -76,7 +71,7 @@ class EthereumLocalReader(
      * parses JSON into Map. But the purpose of further processing and caching for some of the requests we want
      * to have actual data types.
      */
-    fun commonRequests(key: JsonRpcRequest): Mono<ByteArray>? {
+    fun commonRequests(key: JsonRpcRequest): Mono<Pair<ByteArray, String?>>? {
         val method = key.method
         val params = key.params
         return when {
@@ -92,8 +87,8 @@ class EthereumLocalReader(
                 }
                 reader.txByHashAsCont()
                     .read(hash)
-                    .map { it.json!! }
+                    .map { it.data.json!! to it.upstreamId }
-                    .switchIfEmpty { Mono.just(nullValue) }
+                    .switchIfEmpty { Mono.just(nullValue to null) }
             }
             method == "eth_getBlockByHash" -> {
                 if (params.size != 2) {
@@ -109,7 +104,7 @@ class EthereumLocalReader(
                 if (withTx) {
                     null
                 } else {
-                    reader.blocksByIdAsCont().read(hash).map { it.json!! }
+                    reader.blocksByIdAsCont().read(hash).map { it.data.json!! to it.upstreamId }
                 }
             }
             method == "eth_getBlockByNumber" -> {
@@ -127,13 +122,14 @@ class EthereumLocalReader(
                 }
                 reader.receipts()
                     .read(hash)
-                    .switchIfEmpty { Mono.just(nullValue) }
+                    .map { it.data to it.upstreamId }
+                    .switchIfEmpty { Mono.just(nullValue to null) }
             }
             else -> null
         }
     }
 
-    fun getBlockByNumber(params: List<Any?>): Mono<ByteArray>? {
+    fun getBlockByNumber(params: List<Any?>): Mono<Pair<ByteArray, String?>>? {
         if (params.size != 2 || params[0] == null || params[1] == null) {
             throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 2 parameters")
         }
@@ -174,6 +170,6 @@ class EthereumLocalReader(
         }
 
         return reader.blocksByHeightAsCont()
-            .read(number).map { it.json!! }
+            .read(number).map { it.data.json!! to it.upstreamId }
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader.Result
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumLikeMultistream
 import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
 import io.emeraldpay.etherjar.hex.HexData
@@ -31,14 +32,15 @@ import reactor.kotlin.core.publisher.switchIfEmpty
 import java.util.concurrent.TimeUnit
 
 class ProduceLogs(
-    private val receipts: Reader<TxId, ByteArray>
+    private val receipts: Reader<TxId, Result<ByteArray>>
 ) {
 
     companion object {
         private val log = LoggerFactory.getLogger(ProduceLogs::class.java)
     }
 
-    constructor(upstream: EthereumLikeMultistream) : this(upstream.getReader().receipts())
+    constructor(upstream: EthereumLikeMultistream) :
+        this(upstream.getReader().receipts())
 
     private val objectMapper = Global.objectMapper
 
@@ -77,6 +79,7 @@ class ProduceLogs(
                 log.warn("Cannot find receipt for tx ${update.transactionId}")
                 Mono.empty()
             }
+            .map { it.data }
             .flatMapMany { jsonBytes ->
                 // receipt could be a null, like when the original block was replaced, etc.
                 // so just skip it as Flux.empty

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

@@ -35,6 +35,9 @@ class JsonRpcResponse(
 
     constructor(result: ByteArray?, error: JsonRpcError?) : this(result, error, NumberId(0))
 
+    constructor(result: ByteArray?, error: JsonRpcError?, resolvedBy: String?) :
+        this(result, error, NumberId(0), null, resolvedBy)
+
     companion object {
         private val NULL_VALUE = "null".toByteArray()
 

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumCachingReaderSpec.groovy

@@ -69,7 +69,7 @@ class EthereumDirectReaderSpec extends Specification {
         then:
         StepVerifier.create(act)
                 .expectNextMatches { block ->
-                    block.hash.toHexWithPrefix() == hash1
+                    block.data.hash.toHexWithPrefix() == hash1
                 }
                 .expectComplete()
                 .verify(Duration.ofSeconds(1))
@@ -138,7 +138,7 @@ class EthereumDirectReaderSpec extends Specification {
         then:
         StepVerifier.create(act)
                 .expectNextMatches { block ->
-                    block.hash.toHexWithPrefix() == hash1
+                    block.data.hash.toHexWithPrefix() == hash1
                 }
                 .expectComplete()
                 .verify(Duration.ofSeconds(1))
@@ -174,7 +174,7 @@ class EthereumDirectReaderSpec extends Specification {
         then:
         StepVerifier.create(act)
                 .expectNextMatches { block ->
-                    block.hash.toHexWithPrefix() == hash1
+                    block.data.hash.toHexWithPrefix() == hash1
                 }
                 .expectComplete()
                 .verify(Duration.ofSeconds(1))
@@ -208,7 +208,7 @@ class EthereumDirectReaderSpec extends Specification {
         when:
         def act = reader.receiptReader.read(TransactionId.from(hash1))
             .block(Duration.ofSeconds(1))
-            .with { new String(it) }
+            .with { new String(it.data) }
         then:
         act == '{"blockHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","blockNumber":"0x64","transactionHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","logs":[]}'
     }
@@ -245,7 +245,7 @@ class EthereumDirectReaderSpec extends Specification {
         when:
         def act = reader.receiptReader.read(TransactionId.from(hash1))
                 .block(Duration.ofSeconds(1))
-                .with { new String(it) }
+                .with { new String(it.data) }
         then:
         act == '{"blockHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","blockNumber":"0x64","transactionHash":"0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5","logs":[]}'
     }
@@ -302,7 +302,7 @@ class EthereumDirectReaderSpec extends Specification {
             }
         }
         when:
-        def act = reader.balanceReader.read(Address.from(address1))
+        def act = reader.balanceReader.read(Address.from(address1)).map {it.data}
         then:
         StepVerifier.create(act)
                 .expectNext(Wei.from("0x100"))
@@ -334,7 +334,7 @@ class EthereumDirectReaderSpec extends Specification {
             }
         }
         when:
-        def act = reader.balanceReader.read(Address.from(address1))
+        def act = reader.balanceReader.read(Address.from(address1)).map {it.data}
         then:
         StepVerifier.create(act)
                 .expectNext(Wei.from("0x100"))
@@ -379,7 +379,7 @@ class EthereumDirectReaderSpec extends Specification {
         then:
         StepVerifier.create(act)
                 .expectNextMatches { block ->
-                    block.hash.toHexWithPrefix() == hash1
+                    block.data.hash.toHexWithPrefix() == hash1
                 }
                 .expectComplete()
                 .verify(Duration.ofSeconds(1))
@@ -422,7 +422,7 @@ class EthereumDirectReaderSpec extends Specification {
         then:
         StepVerifier.create(act)
                 .expectNextMatches { block ->
-                    block.hash.toHexWithPrefix() == hash1
+                    block.data.hash.toHexWithPrefix() == hash1
                 }
                 .expectComplete()
                 .verify(Duration.ofSeconds(1))

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumLocalReaderSpec.groovy

@@ -69,7 +69,9 @@ class EthereumLocalReaderSpec extends Specification {
             _ * blocksByIdAsCont() >> new EmptyReader<>()
             _ * txByHashAsCont() >> new EmptyReader<>()
             1 * blocksByHeightAsCont() >> Mock(Reader) {
-                1 * read(101L) >> Mono.just(TestingCommons.blockForEthereum(101L))
+                1 * read(101L) >> Mono.just(
+                        new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(101L), null)
+                )
             }
         }
         def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
@@ -81,8 +83,8 @@ class EthereumLocalReaderSpec extends Specification {
         then:
         act != null
         with(act.block()) {
-            it.length > 0
+            it.first.length > 0
-            with(Global.objectMapper.readValue(it, BlockJson)) {
+            with(Global.objectMapper.readValue(it.first, BlockJson)) {
                 number == 101
             }
         }
@@ -95,7 +97,9 @@ class EthereumLocalReaderSpec extends Specification {
             _ * blocksByIdAsCont() >> new EmptyReader<>()
             _ * txByHashAsCont() >> new EmptyReader<>()
             1 * blocksByHeightAsCont() >> Mock(Reader) {
-                1 * read(0L) >> Mono.just(TestingCommons.blockForEthereum(0L))
+                1 * read(0L) >> Mono.just(
+                        new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(0L), null)
+                )
             }
         }
         def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
@@ -107,8 +111,8 @@ class EthereumLocalReaderSpec extends Specification {
         then:
         act != null
         with(act.block()) {
-            it.length > 0
+            it.first.length > 0
-            with(Global.objectMapper.readValue(it, BlockJson)) {
+            with(Global.objectMapper.readValue(it.first, BlockJson)) {
                 number == 0
             }
         }
@@ -121,7 +125,9 @@ class EthereumLocalReaderSpec extends Specification {
             _ * blocksByIdAsCont() >> new EmptyReader<>()
             _ * txByHashAsCont() >> new EmptyReader<>()
             1 * blocksByHeightAsCont() >> Mock(Reader) {
-                1 * read(74735L) >> Mono.just(TestingCommons.blockForEthereum(74735L))
+                1 * read(74735L) >> Mono.just(
+                        new EthereumDirectReader.Result<>(TestingCommons.blockForEthereum(74735L), null)
+                )
             }
         }
         def methods = new DefaultEthereumMethods(Chain.ETHEREUM__MAINNET)
@@ -133,8 +139,8 @@ class EthereumLocalReaderSpec extends Specification {
         then:
         act != null
         with(act.block()) {
-            it.length > 0
+            it.first.length > 0
-            with(Global.objectMapper.readValue(it, BlockJson)) {
+            with(Global.objectMapper.readValue(it.first, BlockJson)) {
                 number == 74735
             }
         }

src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy

@@ -18,6 +18,7 @@ package io.emeraldpay.dshackle.upstream.ethereum.subscribe
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader
 import reactor.core.publisher.Mono
 import spock.lang.Specification
 
@@ -38,7 +39,8 @@ class ProduceLogsSpec extends Specification {
                 '  }'
 
         def receipts = Mock(Reader) {
-            1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(receipt.getBytes())
+            1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >>
+                    Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null))
         }
         def producer = new ProduceLogs(receipts)
         def update = new ConnectBlockUpdates.Update(
@@ -61,7 +63,8 @@ class ProduceLogsSpec extends Specification {
         String receipt = 'null'
 
         def receipts = Mock(Reader) {
-            1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(receipt.getBytes())
+            1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >>
+                    Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null))
         }
         def producer = new ProduceLogs(receipts)
         def update = new ConnectBlockUpdates.Update(
@@ -107,7 +110,8 @@ class ProduceLogsSpec extends Specification {
                 '  }'
 
         def receipts = Mock(Reader) {
-            1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(receipt.getBytes())
+            1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >>
+                    Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null))
         }
         def producer = new ProduceLogs(receipts)
         def update = new ConnectBlockUpdates.Update(
@@ -153,7 +157,8 @@ class ProduceLogsSpec extends Specification {
                 '  }'
 
         def receipts = Mock(Reader) {
-            1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(receipt.getBytes())
+            1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >>
+                    Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null))
         }
         def producer = new ProduceLogs(receipts)
         def update = new ConnectBlockUpdates.Update(
@@ -264,7 +269,8 @@ class ProduceLogsSpec extends Specification {
                 '  }'
 
         def receipts = Mock(Reader) {
-            1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >> Mono.just(receipt.getBytes())
+            1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >>
+                    Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null))
         }
         def producer = new ProduceLogs(receipts)
         def update = new ConnectBlockUpdates.Update(
@@ -343,7 +349,8 @@ class ProduceLogsSpec extends Specification {
                 '  }'
 
         def receipts = Mock(Reader) {
-            1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >> Mono.just(receipt.getBytes())
+            1 * it.read(TxId.from("0xb5e554178a94fd993111f2ae64cb708cb0899d7b5182024e70d5c468164a8bec")) >>
+                    Mono.just(new EthereumDirectReader.Result<>(receipt.getBytes(), null))
         }
         def producer = new ProduceLogs(receipts)
         def update1 = new ConnectBlockUpdates.Update(