solution: Redis cache for transactions

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

@@ -46,7 +46,7 @@ class BlocksRedisCache(
             return Mono.empty()
         }
         return Mono.just(block)
-                .flatMap {
+                .flatMap { block ->
                     val data = objectMapper.writeValueAsString(block)
                     //default caching time is age of the block, i.e. block create hour ago
                     //keep for hour, but block create 10 seconds ago cache for 10 seconds, as it

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

@@ -0,0 +1,84 @@
+package io.emeraldpay.dshackle.cache
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.grpc.Chain
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.rpc.json.BlockJson
+import io.infinitape.etherjar.rpc.json.TransactionJson
+import io.infinitape.etherjar.rpc.json.TransactionRefJson
+import io.lettuce.core.api.reactive.RedisReactiveCommands
+import org.slf4j.LoggerFactory
+import reactor.core.publisher.Mono
+import reactor.util.function.Tuples
+import java.time.Instant
+import java.util.concurrent.TimeUnit
+import kotlin.math.min
+
+/**
+ * Cache transactions in Redis, up to 24 hours.
+ */
+class TxRedisCache(
+        private val redis: RedisReactiveCommands<String, String>,
+        private val chain: Chain,
+        private val objectMapper: ObjectMapper
+): Reader<TransactionId, TransactionJson> {
+
+    companion object {
+        private val log = LoggerFactory.getLogger(TxRedisCache::class.java)
+        // max caching time is 24 hours
+        private const val MAX_CACHE_TIME_HOURS = 24L
+    }
+
+    override fun read(key: TransactionId): Mono<TransactionJson> {
+        return redis.get(key(key))
+                .map { data ->
+                    objectMapper.readValue(data, TransactionJson::class.java) as TransactionJson
+                }.onErrorResume {
+                    Mono.empty()
+                }
+    }
+
+    open fun evict(block: BlockJson<TransactionRefJson>): Mono<Void> {
+        return Mono.just(block)
+                .map { block ->
+                    block.transactions.map {
+                        key(it.hash)
+                    }.toTypedArray()
+                }.flatMap { keys ->
+                    redis.del(*keys)
+                }.then()
+    }
+
+
+    open fun add(tx: TransactionJson, block: BlockJson<TransactionRefJson>): Mono<Void> {
+        if (tx.blockHash == null || block.hash == null || tx.blockHash != block.hash || block.timestamp == null) {
+            return Mono.empty()
+        }
+        return Mono.just(Tuples.of(tx, block))
+                .flatMap {
+                    val data = objectMapper.writeValueAsString(it.t1)
+                    //default caching time is age of the block, i.e. block create hour ago
+                    //keep for hour, but block create 10 seconds ago cache for 10 seconds, as it
+                    //still can be replaced in the blockchain
+                    val age = Instant.now().epochSecond - it.t2.timestamp.epochSecond
+                    val ttl = min(age, TimeUnit.HOURS.toSeconds(MAX_CACHE_TIME_HOURS))
+                    redis.setex(key(it.t1.hash), ttl, data)
+                }
+                .doOnError {
+                    log.warn("Failed to save to Redis: ${it.message}")
+                }
+                //if failed to cache, just continue without it
+                .onErrorResume {
+                    Mono.empty()
+                }
+                .then()
+    }
+
+    /**
+     * Key in Redis
+     */
+    open fun key(hash: TransactionId): String {
+        return "tx:${chain.id}:${hash.toHex()}"
+    }
+}

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

@@ -0,0 +1,111 @@
+package io.emeraldpay.dshackle.cache
+
+import io.emeraldpay.dshackle.test.TestingCommons
+import io.emeraldpay.grpc.Chain
+import io.infinitape.etherjar.domain.BlockHash
+import io.infinitape.etherjar.domain.TransactionId
+import io.infinitape.etherjar.domain.Wei
+import io.infinitape.etherjar.rpc.json.BlockJson
+import io.infinitape.etherjar.rpc.json.TransactionJson
+import io.infinitape.etherjar.rpc.json.TransactionRefJson
+import io.lettuce.core.RedisClient
+import io.lettuce.core.api.StatefulRedisConnection
+import spock.lang.Specification
+
+import java.time.Instant
+import java.time.temporal.ChronoUnit
+
+class TxRedisCacheSpec extends Specification {
+
+    String hash1 = "0xd3f34def3c56ba4e701540d15edaff9acd2a1c968a7ff83b3300ab5dfd5f6aab"
+    String hash2 = "0x4aabdaff9acd2f30d15e00ab5dfd5f6c56ba4ea1c968a7ff8d3f34de70153b33"
+    String hash3 = "0x40d15edaff9acdabd2a1c96fd5f683b3300aad34e7015f34def3c56ba8a7ffb5"
+    String hash4 = "0xa4e7a75dfd5f6a83b3304dc56bfa0abfd3fef01540d15edafc9683f9acd2a13b"
+    TxRedisCache cache
+
+    def setup() {
+        RedisClient client = RedisClient.create("redis://localhost:6379");
+        StatefulRedisConnection<String, String> connection = client.connect();
+        connection.sync().flushdb()
+        StatefulRedisConnection<String, String> redis = connection
+        cache = new TxRedisCache(redis.reactive(), Chain.ETHEREUM, TestingCommons.objectMapper())
+    }
+
+    def "Add and read"() {
+        setup:
+        def block = new BlockJson<TransactionRefJson>()
+        block.number = 100
+        block.timestamp = Instant.now().minusSeconds(100).truncatedTo(ChronoUnit.SECONDS)
+        block.hash = BlockHash.from(hash1)
+        block.transactions = []
+        block.uncles = []
+
+        def tx = new TransactionJson()
+        tx.hash = TransactionId.from(hash1)
+        tx.blockHash = block.hash
+        tx.blockNumber = block.number
+        tx.value = Wei.ofEthers(1.234)
+        tx.nonce = 0
+
+        when:
+        cache.add(tx, block).subscribe()
+        def act = cache.read(TransactionId.from(hash1)).block()
+        then:
+        act == tx
+    }
+
+    def "Evict all by block data"() {
+        when:
+        def block1 = new BlockJson<TransactionRefJson>()
+        block1.hash = BlockHash.from(hash1)
+        block1.number = 100
+        block1.timestamp = Instant.now().minusSeconds(100).truncatedTo(ChronoUnit.SECONDS)
+        block1.transactions = [
+                new TransactionRefJson(TransactionId.from(hash1)),
+                new TransactionRefJson(TransactionId.from(hash2)),
+        ]
+        def block2 = new BlockJson<TransactionRefJson>()
+        block2.hash = BlockHash.from(hash2)
+        block2.number = 101
+        block2.timestamp = Instant.now().minusSeconds(100).truncatedTo(ChronoUnit.SECONDS)
+        block2.transactions = [
+                new TransactionRefJson(TransactionId.from(hash3)),
+                new TransactionRefJson(TransactionId.from(hash4)),
+        ]
+
+        [hash1, hash2].eachWithIndex{ String hash, int i ->
+            def tx = new TransactionJson()
+            tx.blockNumber = block1.number
+            tx.blockHash = block1.hash
+            tx.hash = TransactionId.from(hash)
+            tx.value = Wei.ofEthers(i)
+            tx.nonce = 0
+            cache.add(tx, block1).subscribe()
+        }
+        [hash3, hash4].eachWithIndex{ String hash, int i ->
+            def tx = new TransactionJson()
+            tx.blockNumber = block2.number
+            tx.blockHash = block2.hash
+            tx.hash = TransactionId.from(hash)
+            tx.value = Wei.ofEthers(i)
+            tx.nonce = 0
+            cache.add(tx, block2).subscribe()
+        }
+
+
+        cache.evict(block1).subscribe()
+
+        def act1 = cache.read(TransactionId.from(hash1)).block()
+        def act2 = cache.read(TransactionId.from(hash2)).block()
+        def act3 = cache.read(TransactionId.from(hash3)).block()
+        def act4 = cache.read(TransactionId.from(hash4)).block()
+
+        then:
+        act1 == null
+        act2 == null
+        act3 != null
+        act3.hash.toHex() == hash3
+        act4 != null
+        act4.hash.toHex() == hash4
+    }
+}