problem: doesn't cache recent Tx Receipts

src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt

@@ -33,6 +33,7 @@ open class Caches(
         private val memBlocksByHash: BlocksMemCache,
         private val blocksByHeight: HeightCache,
         private val memTxsByHash: TxMemCache,
+        private val memReceipts: ReceiptMemCache,
         private val redisBlocksByHash: BlocksRedisCache?,
         private val redisTxsByHash: TxRedisCache?,
         private val redisReceipts: ReceiptRedisCache?,
@@ -59,6 +60,8 @@ open class Caches(
     private val txsByHash: Reader<TxId, TxContainer>
     private val receiptByHash: Reader<TxId, ByteArray>
 
+    private var head: Head? = null
+
     init {
         blocksByHash = if (redisBlocksByHash == null) {
             memBlocksByHash
@@ -70,26 +73,24 @@ open class Caches(
         } else {
             CompoundReader(memTxsByHash, redisTxsByHash)
         }
-        receiptByHash = redisReceipts ?: EmptyReader()
+        receiptByHash = if (redisReceipts == null) {
+            memReceipts
+        } else {
+            CompoundReader(memReceipts, redisReceipts)
+        }
     }
 
     fun setHead(head: Head) {
+        this.head = head
         redisTxsByHash?.head = head
         redisReceipts?.head = head
     }
 
-    /**
-     * Cache data that was just requested
-     */
-    fun cacheRequested(data: Any) {
-        if (data is TxContainer) {
-            cache(Tag.REQUESTED, data)
-        } else if (data is BlockContainer) {
-            cache(Tag.REQUESTED, data)
-        }
-    }
-
     open fun cacheReceipt(tag: Tag, data: DefaultContainer<TransactionReceiptJson>) {
+        val currentHeight = head?.getCurrentHeight()
+        if (currentHeight != null && data.height != null && memReceipts.acceptsRecentBlocks(currentHeight - data.height)) {
+            memReceipts.add(data)
+        }
         //TODO move subscription to the caller
         redisReceipts?.add(data)?.subscribe()
     }
@@ -165,6 +166,7 @@ open class Caches(
         memBlocksByHash.get(blockId)?.let { block ->
             memTxsByHash.evict(block)
             redisTxsByHash?.evict(block)
+            memReceipts.evict(block)
             evicted = true
         }
         if (!evicted) {
@@ -224,6 +226,7 @@ open class Caches(
         private var blocksByHash: BlocksMemCache? = null
         private var blocksByHeight: HeightCache? = null
         private var txsByHash: TxMemCache? = null
+        private var receipts: ReceiptMemCache? = null
         private var redisBlocksByHash: BlocksRedisCache? = null
         private var redisTxsByHash: TxRedisCache? = null
         private var redisReceiptCache: ReceiptRedisCache? = null
@@ -259,6 +262,11 @@ open class Caches(
             return this
         }
 
+        fun setReceipts(cache: ReceiptMemCache): Builder {
+            this.receipts = cache
+            return this
+        }
+
         fun setHeightByHash(cache: HeightByHashRedisCache): Builder {
             redisHeightByHashCache = cache
             return this
@@ -274,7 +282,10 @@ open class Caches(
             if (txsByHash == null) {
                 txsByHash = TxMemCache()
             }
-            return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!,
+            if (receipts == null) {
+                receipts = ReceiptMemCache()
+            }
+            return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, receipts!!,
                     redisBlocksByHash, redisTxsByHash, redisReceiptCache, redisHeightByHashCache)
         }
     }

src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptMemCache.kt

@@ -0,0 +1,64 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.cache
+
+import com.github.benmanes.caffeine.cache.Caffeine
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.DefaultContainer
+import io.emeraldpay.dshackle.data.TxId
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+
+/**
+ * Keeps receipts for recent blocks in memory
+ */
+open class ReceiptMemCache(
+        // how many blocks to keeps in memory
+        val blocks: Int = 6
+) : Reader<TxId, ByteArray> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(ReceiptMemCache::class.java)
+    }
+
+    private val mapping = Caffeine.newBuilder()
+            .maximumSize(blocks * 200L)
+            .build<TxId, ByteArray>()
+
+    open fun evict(block: BlockContainer) {
+        block.transactions.forEach {
+            mapping.invalidate(it)
+        }
+    }
+
+    override fun read(key: TxId): Mono<ByteArray> {
+        return mapping.getIfPresent(key)?.let { Mono.just(it) } ?: Mono.empty()
+    }
+
+    open fun add(receipt: DefaultContainer<TransactionReceiptJson>): Mono<Void> {
+        if (receipt.txId != null && receipt.json != null) {
+            mapping.put(receipt.txId, receipt.json)
+        }
+        return Mono.empty()
+    }
+
+    open fun acceptsRecentBlocks(heightDelta: Long): Boolean {
+        return blocks <= heightDelta && heightDelta >= 0
+    }
+
+}

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

@@ -200,12 +200,6 @@ open class NativeCall(
                 .map {
                     CallResult(ctx.id, it.value, null)
                 }
-                .doOnNext {
-                    it.result?.let { value ->
-                        ctx.upstream.postprocessor
-                                .onReceive(ctx.payload.method, ctx.payload.params, value)
-                    }
-                }
                 .onErrorResume { t ->
                     val failure = if (t is CallFailure) {
                         CallResult.fail(t.id, t.reason)

src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt

@@ -140,6 +140,7 @@ abstract class Multistream(
         apis.request(1)
         return Mono.from(apis)
                 .map(Upstream::getApi)
+                .map { RequestPostprocessor.wrap(it, postprocessor) } //TODO do it on upstream init, not each time it's called
                 .switchIfEmpty(Mono.error(Exception("No API available for $chain")))
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt

@@ -1,10 +1,38 @@
 package io.emeraldpay.dshackle.upstream
 
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import reactor.core.publisher.Mono
+
 interface RequestPostprocessor {
 
-    fun onReceive(method: String, params: List<Any>, json: ByteArray)
+    fun onReceive(method: String, params: List<Any?>, json: ByteArray)
 
     class Empty : RequestPostprocessor {
-        override fun onReceive(method: String, params: List<Any>, json: ByteArray) {}
+        override fun onReceive(method: String, params: List<Any?>, json: ByteArray) {}
+    }
+
+    companion object {
+        fun wrap(reader: Reader<JsonRpcRequest, JsonRpcResponse>, processor: RequestPostprocessor): Reader<JsonRpcRequest, JsonRpcResponse> {
+            return Wrapper(reader, processor)
+        }
+    }
+
+    class Wrapper(
+            private val reader: Reader<JsonRpcRequest, JsonRpcResponse>,
+            private val processor: RequestPostprocessor
+    ) : Reader<JsonRpcRequest, JsonRpcResponse> {
+
+        override fun read(key: JsonRpcRequest): Mono<JsonRpcResponse> {
+            return reader.read(key)
+                    .doOnNext {
+                        if (it.hasResult()) {
+                            val result = it.getResult()
+                            processor.onReceive(key.method, key.params, result)
+                        }
+                    }
+        }
+
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/Selector.kt

@@ -26,6 +26,7 @@ class Selector {
 
     companion object {
 
+        @JvmStatic
         val empty = EmptyMatcher()
 
         @JvmStatic

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

@@ -20,6 +20,7 @@ import io.emeraldpay.dshackle.cache.Caches
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.DefaultContainer
 import io.emeraldpay.dshackle.data.TxId
+import io.emeraldpay.dshackle.upstream.Head
 import io.emeraldpay.dshackle.upstream.RequestPostprocessor
 import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
 import org.slf4j.LoggerFactory
@@ -32,7 +33,7 @@ class CacheRequested(
         private val log = LoggerFactory.getLogger(CacheRequested::class.java)
     }
 
-    override fun onReceive(method: String, params: List<Any>, json: ByteArray) {
+    override fun onReceive(method: String, params: List<Any?>, json: ByteArray) {
         try {
             if (method == "eth_getTransactionReceipt") {
                 cacheTxReceipt(params, json)
@@ -42,7 +43,7 @@ class CacheRequested(
         }
     }
 
-    fun cacheTxReceipt(params: List<Any>, json: ByteArray) {
+    fun cacheTxReceipt(params: List<Any?>, json: ByteArray) {
         if (params.size != 1) {
             return
         }

src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy

@@ -19,12 +19,17 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.data.DefaultContainer
 import io.emeraldpay.dshackle.data.TxContainer
+import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.Head
+import io.emeraldpay.etherjar.domain.Address
 import io.emeraldpay.etherjar.domain.BlockHash
 import io.emeraldpay.etherjar.domain.TransactionId
 import io.emeraldpay.etherjar.rpc.json.BlockJson
 import io.emeraldpay.etherjar.rpc.json.TransactionJson
+import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
 import io.emeraldpay.etherjar.rpc.json.TransactionRefJson
 import reactor.core.publisher.Mono
 import spock.lang.Specification
@@ -227,4 +232,36 @@ class CachesSpec extends Specification {
         1 * blocksCache.read(block.hash) >> Mono.just(block)
         1 * txRedisCache.add(TxContainer.from(tx1), block) >> Mono.just(1).then()
     }
+
+    def "Put receipt into mem cache"() {
+        setup:
+        def receipt = new TransactionReceiptJson().tap {
+            transactionHash = TransactionId.from("0xc7529e79f78f58125abafeaea01fe3abdc6f45c173d5dfb36716cbc526e5b2d1")
+            blockHash = BlockHash.from("0x48249c81bfced2e6fe2536126471b73d83c4f21de75f88a16feb57cc566b991b")
+            blockNumber = 0xccf6e2
+            from = Address.from("0x3a1428354c99b119d891a30d326bad92e36e896a")
+            logs = []
+        }
+        def receiptContainer = new DefaultContainer(
+                TxId.from(receipt.transactionHash),
+                BlockId.from(receipt.blockHash),
+                receipt.blockNumber,
+                Global.objectMapper.writeValueAsBytes(receipt),
+                receipt
+        )
+
+        ReceiptMemCache receiptMemCache = Mock()
+        def caches = Caches.newBuilder()
+                .setReceipts(receiptMemCache)
+                .build()
+        Head head = Mock()
+        caches.setHead(head)
+        when:
+        caches.cacheReceipt(Caches.Tag.REQUESTED, receiptContainer)
+
+        then:
+        1 * head.getCurrentHeight() >> 0xccf6e2
+        1 * receiptMemCache.acceptsRecentBlocks(0) >> true
+        1 * receiptMemCache.add(receiptContainer)
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptMemCacheSpec.groovy

@@ -0,0 +1,100 @@
+/**
+ * Copyright (c) 2021 EmeraldPay, Inc
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.emeraldpay.dshackle.cache
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.Global
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.data.BlockId
+import io.emeraldpay.dshackle.data.DefaultContainer
+import io.emeraldpay.dshackle.data.TxId
+import io.emeraldpay.etherjar.domain.Address
+import io.emeraldpay.etherjar.domain.BlockHash
+import io.emeraldpay.etherjar.domain.TransactionId
+import io.emeraldpay.etherjar.rpc.json.TransactionReceiptJson
+import spock.lang.Specification
+
+import java.time.Instant
+
+class ReceiptMemCacheSpec extends Specification {
+
+    ObjectMapper objectMapper = Global.objectMapper
+
+    def "Add and read"() {
+        setup:
+        def cache = new ReceiptMemCache()
+
+        def receipt = new TransactionReceiptJson().tap {
+            transactionHash = TransactionId.from("0xc7529e79f78f58125abafeaea01fe3abdc6f45c173d5dfb36716cbc526e5b2d1")
+            blockHash = BlockHash.from("0x48249c81bfced2e6fe2536126471b73d83c4f21de75f88a16feb57cc566b991b")
+            blockNumber = 0xccf6e2
+            from = Address.from("0x3a1428354c99b119d891a30d326bad92e36e896a")
+            logs = []
+        }
+        def receiptContainer = new DefaultContainer(
+                TxId.from(receipt.transactionHash),
+                BlockId.from(receipt.blockHash),
+                receipt.blockNumber,
+                objectMapper.writeValueAsBytes(receipt),
+                receipt
+        )
+
+        when:
+        cache.add(receiptContainer)
+        def act = cache.read(TxId.from(receipt.transactionHash)).block()
+        then:
+        act != null
+        objectMapper.readValue(act, TransactionReceiptJson.class) == receipt
+    }
+
+    def "Evict by block"() {
+        setup:
+        def cache = new ReceiptMemCache()
+
+        def receipt = new TransactionReceiptJson().tap {
+            transactionHash = TransactionId.from("0xc7529e79f78f58125abafeaea01fe3abdc6f45c173d5dfb36716cbc526e5b2d1")
+            blockHash = BlockHash.from("0x48249c81bfced2e6fe2536126471b73d83c4f21de75f88a16feb57cc566b991b")
+            blockNumber = 0xccf6e2
+            from = Address.from("0x3a1428354c99b119d891a30d326bad92e36e896a")
+            logs = []
+        }
+        def receiptContainer = new DefaultContainer(
+                TxId.from(receipt.transactionHash),
+                BlockId.from(receipt.blockHash),
+                receipt.blockNumber,
+                objectMapper.writeValueAsBytes(receipt),
+                receipt
+        )
+
+        def blockContainer = new BlockContainer(
+                receipt.blockNumber, BlockId.from(receipt.blockHash),
+                BigInteger.ONE,
+                Instant.now(),
+                false,
+                "{}".bytes,
+                null,
+                [TxId.from(receipt.transactionHash)]
+        )
+
+        when:
+        cache.add(receiptContainer)
+        cache.evict(blockContainer)
+        def act = cache.read(TxId.from(receipt.transactionHash)).block()
+        then:
+        act == null
+    }
+
+}

src/test/groovy/io/emeraldpay/dshackle/upstream/MultistreamSpec.groovy

@@ -17,12 +17,18 @@
 package io.emeraldpay.dshackle.upstream
 
 import io.emeraldpay.dshackle.cache.Caches
+import io.emeraldpay.dshackle.config.UpstreamsConfig
 import io.emeraldpay.dshackle.quorum.AlwaysQuorum
+import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.test.EthereumUpstreamMock
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumMultistream
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
 import io.emeraldpay.grpc.Chain
+import org.jetbrains.annotations.NotNull
+import reactor.core.publisher.Mono
 import spock.lang.Specification
 
 import java.time.Duration
@@ -163,4 +169,61 @@ class MultistreamSpec extends Specification {
         then:
         !act
     }
+
+    def "Call postprocess after api use"() {
+        setup:
+        def request = new JsonRpcRequest("test_foo", [1], 1)
+
+        def api = TestingCommons.api()
+        api.answer("test_foo", [1], "test")
+        def postprocessor = Mock(RequestPostprocessor)
+        def up = TestingCommons.upstream(api)
+        def multistream = new TestMultistream([up], postprocessor)
+
+        when:
+        def rdr = multistream.getDirectApi(Selector.empty).block(Duration.ofSeconds(1))
+        def act = rdr.read(request).block(Duration.ofSeconds(1))
+
+        then:
+        act != null
+        act.hasResult()
+        act.resultAsProcessedString == "test"
+        1 * postprocessor.onReceive("test_foo", [1], "\"test\"".bytes)
+    }
+
+    class TestMultistream extends Multistream {
+
+        TestMultistream(List<Upstream> upstreams, @NotNull RequestPostprocessor postprocessor) {
+            super(Chain.ETHEREUM, upstreams, Caches.default(), postprocessor)
+        }
+
+        @Override
+        Mono<Reader<JsonRpcRequest, JsonRpcResponse>> getRoutedApi(@NotNull Selector.Matcher matcher) {
+            return null
+        }
+
+        @Override
+        Head updateHead() {
+            return null
+        }
+
+        @Override
+        void setHead(@NotNull Head head) {
+
+        }
+
+        @Override
+        Head getHead() {
+            return null
+        }
+
+        @Override
+        Collection<UpstreamsConfig.Labels> getLabels() {
+            return null
+        }
+
+        public <T extends Upstream> T cast(Class<T> selfType) {
+            return this
+        }
+    }
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/RequestPostprocessorSpec.groovy

@@ -0,0 +1,47 @@
+package io.emeraldpay.dshackle.upstream
+
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
+import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
+import reactor.core.publisher.Mono
+import spock.lang.Specification
+
+import java.time.Duration
+
+class RequestPostprocessorSpec extends Specification {
+
+    def "Wrappers calls onReceive for a value"() {
+        setup:
+        def request = new JsonRpcRequest("test_foo", [1], 1)
+        def processor = Mock(RequestPostprocessor)
+        def api = TestingCommons.api()
+        api.answer("test_foo", [1], "test")
+        def wrapped = new RequestPostprocessor.Wrapper(api, processor)
+
+        when:
+        def act = wrapped.read(request).block(Duration.ofSeconds(1))
+
+        then:
+        act.hasResult()
+        act.resultAsProcessedString == "test"
+        1 * processor.onReceive("test_foo", [1], "\"test\"".bytes)
+    }
+
+    def "Wrappers doesn't call onReceive for no value"() {
+        setup:
+        def request = new JsonRpcRequest("test_foo", [1], 1)
+        def processor = Mock(RequestPostprocessor)
+        Reader<JsonRpcRequest, JsonRpcResponse> reader = Mock(Reader) {
+            1 * it.read(request) >> Mono.empty()
+        }
+        def wrapped = new RequestPostprocessor.Wrapper(reader, processor)
+
+        when:
+        def act = wrapped.read(request).block(Duration.ofSeconds(1))
+
+        then:
+        act == null
+        0 * processor.onReceive(_, _, _)
+    }
+}