Merge pull request #41 from emeraldpay/fix/block-with-tx

2020-07-20 21:41:40 -04:00
parent 9d752fc8f1 c80eb14847
commit 74312e1e3f
15 changed files with 381 additions and 134 deletions
--- a/build.gradle
+++ b/build.gradle
@@ -104,6 +104,7 @@ dependencies {
    implementation 'org.apache.commons:commons-collections4:4.3'
    implementation 'javax.annotation:javax.annotation-api:1.3.2'
    implementation 'org.bouncycastle:bcprov-jdk15on:1.61'
+    implementation 'com.github.ben-manes.caffeine:caffeine:2.8.5'

    implementation "org.slf4j:slf4j-api:$slf4jVersion"
    implementation "org.apache.logging.log4j:log4j-slf4j-impl:2.11.1"
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksMemCache.kt
@@ -16,36 +16,33 @@
 */
 package io.emeraldpay.dshackle.cache

+import com.github.benmanes.caffeine.cache.Caffeine
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.reader.Reader
 import reactor.core.publisher.Mono
-import java.util.concurrent.ConcurrentHashMap
-import java.util.concurrent.ConcurrentLinkedQueue

 open class BlocksMemCache(
-        val maxSize: Int = 64
+        maxSize: Int = 64
 ) : Reader<BlockId, BlockContainer> {

-    private val mapping = ConcurrentHashMap<BlockId, BlockContainer>()
-    private val queue = ConcurrentLinkedQueue<BlockId>()
+    private val mapping = Caffeine.newBuilder()
+            .maximumSize(maxSize.toLong())
+            .build<BlockId, BlockContainer>()

    override fun read(key: BlockId): Mono<BlockContainer> {
-        return Mono.justOrEmpty(mapping[key])
+        return Mono.justOrEmpty(get(key))
    }

    open fun get(key: BlockId): BlockContainer? {
-        return mapping[key]
+        return mapping.getIfPresent(key)
    }

    open fun add(block: BlockContainer) {
        mapping.put(block.hash, block)
-        queue.add(block.hash)
-
-        while (queue.size > maxSize) {
-            val old = queue.remove()
-            mapping.remove(old)
-        }
    }

+    open fun purge() {
+        mapping.cleanUp()
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt
@@ -97,30 +97,17 @@ open class Caches(
            return
        }
        memTxsByHash.add(tx)
-        memBlocksByHash.get(tx.blockId)?.let { block ->
-            redisTxsByHash?.add(tx, block)
-        }
+        //TODO move subscription to the caller
+        getBlocksByHash().read(tx.blockId).flatMap { block ->
+            redisTxsByHash?.add(tx, block) ?: Mono.empty()
+        }.subscribe()
    }

    fun cache(tag: Tag, block: BlockContainer) {
        val job = ArrayList<Mono<Void>>()
        if (tag == Tag.LATEST) {
-            //for LATEST data cache in memory, it will be short living so better to avoid Redis
-            memBlocksByHash.add(block)
-            val replaced = blocksByHeight.add(block)
-            //evict cached transactions if an existing block was updated
-            replaced?.let { replacedBlockHash ->
-                var evicted = false
-                redisBlocksByHash?.evict(replacedBlockHash)
-                memBlocksByHash.get(replacedBlockHash)?.let { block ->
-                    memTxsByHash.evict(block)
-                    redisTxsByHash?.evict(block)
-                    evicted = true
-                }
-                if (!evicted) {
-                    memTxsByHash.evict(replacedBlockHash)
-                }
-            }
+            //for LATEST data cache it in memory, it may be short living so better to avoid Redis
+            memoizeBlock(block)
        } else if (tag == Tag.REQUESTED) {
            var blockOnlyContainer: BlockContainer? = null
            var jsonValue: BlockJson<*>? = null
@@ -132,6 +119,7 @@ open class Caches(
            } else {
                blockOnlyContainer = block
            }
+            memoizeBlock(blockOnlyContainer)
            memBlocksByHash.add(blockOnlyContainer)
            redisBlocksByHash?.add(blockOnlyContainer)?.let(job::add)

@@ -142,11 +130,11 @@ open class Caches(
                    val transactions = plainTransactions.map { tx ->
                        TxContainer.from(tx)
                    }
-                    transactions.forEach {
-                        cache(Tag.REQUESTED, it)
-                    }
                    if (redisTxsByHash != null) {
-                        job.add(Flux.fromIterable(transactions).flatMap { redisTxsByHash.add(it, block) }.then())
+                        job.add(Flux.fromIterable(transactions)
+                                .doOnNext { memTxsByHash.add(it) }
+                                .flatMap { redisTxsByHash.add(it, block) }
+                                .then())
                    }
                }
            }
@@ -154,6 +142,29 @@ open class Caches(
        Flux.fromIterable(job).flatMap { it }.subscribe() //TODO move out to a caller
    }

+    /**
+     * Cache the block only in memory
+     */
+    fun memoizeBlock(block: BlockContainer) {
+        memBlocksByHash.add(block)
+        val replaced = blocksByHeight.add(block)
+        //evict cached transactions if an existing block was updated
+        replaced?.let { evict(it) }
+    }
+
+    fun evict(blockId: BlockId) {
+        var evicted = false
+        redisBlocksByHash?.evict(blockId)
+        memBlocksByHash.get(blockId)?.let { block ->
+            memTxsByHash.evict(block)
+            redisTxsByHash?.evict(block)
+            evicted = true
+        }
+        if (!evicted) {
+            memTxsByHash.evict(blockId)
+        }
+    }
+
    fun getBlocksByHash(): Reader<BlockId, BlockContainer> {
        return blocksByHash
    }
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightCache.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/HeightCache.kt
@@ -15,41 +15,35 @@
 */
 package io.emeraldpay.dshackle.cache

+import com.github.benmanes.caffeine.cache.Caffeine
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.reader.Reader
-import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
-import java.util.concurrent.ConcurrentHashMap

 /**
 * Memory cache for blocks heights, keeps mapping height->hash.
 */
 open class HeightCache(
-        val maxSize: Int = 256
+        maxSize: Int = 512
 ) : Reader<Long, BlockId> {

-    companion object {
-        private val log = LoggerFactory.getLogger(HeightCache::class.java)
-    }
-
-    private val heights = ConcurrentHashMap<Long, BlockId>()
+    private val heights = Caffeine.newBuilder()
+            .maximumSize(maxSize.toLong())
+            .build<Long, BlockId>()

    override fun read(key: Long): Mono<BlockId> {
-        return Mono.justOrEmpty(heights[key])
+        return Mono.justOrEmpty(heights.getIfPresent(key))
    }

    open fun add(block: BlockContainer): BlockId? {
-        val existing = heights[block.height]
-        heights[block.height] = block.hash
-
-        // evict old numbers if full
-        var dropHeight = block.height - maxSize
-        while (heights.size > maxSize && dropHeight < block.height) {
-            heights.remove(dropHeight)
-            dropHeight++
-        }
-
-        return existing
+        val previousId = heights.getIfPresent(block.height)
+        heights.put(block.height, block.hash)
+        return previousId
    }
+
+    fun purge() {
+        heights.cleanUp()
+    }
+
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/OnTxRedisCache.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/OnTxRedisCache.kt
@@ -117,7 +117,7 @@ abstract class OnTxRedisCache<T>(
                }
    }

-    fun add(id: TxId, value: T, block: BlockContainer?, blockHeight: Long?): Mono<Void> {
+    open fun add(id: TxId, value: T, block: BlockContainer?, blockHeight: Long?): Mono<Void> {
        return Mono.just(id)
                .flatMap {
                    val key = key(it)
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxMemCache.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxMemCache.kt
@@ -15,6 +15,7 @@
 */
 package io.emeraldpay.dshackle.cache

+import com.github.benmanes.caffeine.cache.Caffeine
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.TxContainer
@@ -22,8 +23,6 @@ import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.reader.Reader
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Mono
-import java.util.concurrent.ConcurrentHashMap
-import java.util.concurrent.ConcurrentLinkedQueue

 /**
 * Memory cache for transactions
@@ -37,24 +36,25 @@ open class TxMemCache(
        private val log = LoggerFactory.getLogger(TxMemCache::class.java)
    }

-    private val mapping = ConcurrentHashMap<TxId, TxContainer>()
-    private val queue = ConcurrentLinkedQueue<TxId>()
+    private val mapping = Caffeine.newBuilder()
+            .maximumSize(maxSize.toLong())
+            .build<TxId, TxContainer>()

    override fun read(key: TxId): Mono<TxContainer> {
-        return Mono.justOrEmpty(mapping[key])
+        return Mono.justOrEmpty(mapping.getIfPresent(key))
    }

    open fun evict(block: BlockContainer) {
        block.transactions.forEach {
-            mapping.remove(it)
+            mapping.invalidate(it)
        }
    }

    open fun evict(block: BlockId) {
-        val ids = mapping.filter { it.value.blockId == block }
-        ids.forEach {
-            mapping.remove(it.key)
-        }
+        val ids = mapping.asMap()
+                .filter { it.value.blockId == block }
+                .map { it.key }
+        mapping.invalidateAll(ids)
    }

    open fun add(tx: TxContainer) {
@@ -63,12 +63,9 @@ open class TxMemCache(
            return
        }
        mapping.put(tx.hash, tx)
-        queue.add(tx.hash)
-
-        while (queue.size > maxSize) {
-            val old = queue.remove()
-            mapping.remove(old)
-        }
    }

+    open fun purge() {
+        mapping.cleanUp()
+    }
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt
@@ -35,7 +35,7 @@ import kotlin.math.min
 /**
 * Cache transactions in Redis, up to 24 hours.
 */
-class TxRedisCache(
+open class TxRedisCache(
        private val redis: RedisReactiveCommands<String, ByteArray>,
        private val chain: Chain
 ) : Reader<TxId, TxContainer>,
@@ -74,7 +74,7 @@ class TxRedisCache(
        )
    }

-    fun add(tx: TxContainer, block: BlockContainer): Mono<Void> {
+    open fun add(tx: TxContainer, block: BlockContainer): Mono<Void> {
        return super.add(tx.hash, tx, block, tx.height)
    }

--- a/src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt
@@ -28,6 +28,12 @@ class TxContainer(
 ) : SourceContainer(json, parsed) {

    companion object {
+        @JvmStatic
+        fun from(raw: ByteArray): TxContainer {
+            val tx = Global.objectMapper.readValue(raw, TransactionJson::class.java)
+            return from(tx, raw)
+        }
+
        @JvmStatic
        fun from(tx: TransactionJson): TxContainer {
            return from(tx, Global.objectMapper.writeValueAsBytes(tx))
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReader.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReader.kt
@@ -15,20 +15,18 @@
 */
 package io.emeraldpay.dshackle.upstream.ethereum

-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.TxContainer
 import io.emeraldpay.dshackle.data.TxId
 import io.emeraldpay.dshackle.reader.Reader
-import io.infinitape.etherjar.rpc.json.BlockJson
-import io.infinitape.etherjar.rpc.json.TransactionJson
-import io.infinitape.etherjar.rpc.json.TransactionRefJson
 import org.slf4j.LoggerFactory
-import org.springframework.beans.BeanUtils
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.util.function.Tuple2
+import reactor.util.function.Tuples
+import java.nio.ByteBuffer
+import java.util.function.BiFunction

 /**
 * Reads blocks with full transactions details. Based on data contained in readers for blocks
@@ -46,40 +44,91 @@ class EthereumFullBlocksReader(
        private val log = LoggerFactory.getLogger(EthereumFullBlocksReader::class.java)
    }

-    override fun read(key: BlockId): Mono<BlockContainer> {
-        return blocks.read(key).flatMap { block ->
-            val block = Global.objectMapper.readValue(block.json, BlockJson::class.java) as BlockJson<TransactionRefJson>
-            val fullBlock = if (block.transactions == null || block.transactions.isEmpty()) {
-                // in fact it's not necessary to create a copy, made just for code clarity but it may be a performance loss
-                val fullBlock = BlockJson<TransactionJson>()
-                BeanUtils.copyProperties(block, fullBlock)
-                Mono.just(fullBlock)
-            } else {
-                Flux.fromIterable(block.transactions)
-                        .map { TxId.from(it.hash) }
-                        .flatMap { txes.read(it) }
-                        .collectList()
-                        .flatMap { list ->
-                            if (block.transactions.size != list.size) {
-                                Mono.empty<BlockJson<TransactionJson>>()
-                            } else {
-                                val fullBlock = BlockJson<TransactionJson>()
-                                BeanUtils.copyProperties(block, fullBlock)
-                                fullBlock.transactions = list.map {
-                                    Global.objectMapper.readValue(it.json, TransactionJson::class.java)
-                                }
-                                Mono.just(fullBlock)
-                            }
-                        }
-            }
-            fullBlock
-                    .map { block ->
-                        BlockContainer(block.number, BlockId.from(block.hash), block.totalDifficulty, block.timestamp, true,
-                                Global.objectMapper.writeValueAsBytes(block),
-                                block.transactions.map { tx -> TxId.from(tx) }
-                        )
-                    }
+    private val accumulate: BiFunction<ByteBuffer, ByteArray, ByteBuffer> = BiFunction { buf, x ->
+        if (buf.remaining() < x.size) {
+            val resize = ByteBuffer.allocate(buf.capacity() + buf.capacity() / 4 + x.size)
+            resize.put(buf.flip()).put(x)
+        } else {
+            buf.put(x)
        }
    }

+    override fun read(key: BlockId): Mono<BlockContainer> {
+        return blocks.read(key).flatMap { block ->
+            if (block.transactions.isEmpty()) {
+                // in fact it's not necessary to create a copy, made just for code clarity but it may be a performance loss
+                val fullBlock = BlockContainer(
+                        block.height, block.hash, block.difficulty, block.timestamp,
+                        true,
+                        block.json,
+                        block.parsed,
+                        block.transactions
+                )
+                return@flatMap Mono.just(fullBlock)
+            }
+
+            val blockSplit = splitByTransactions(block.json!!)
+
+            val transactions = Flux.fromIterable(block.transactions)
+                    .flatMap { txes.read(it) }
+                    .collectList()
+
+            return@flatMap transactions.flatMap { transactionsData ->
+                // make sure that all transaction are loaded, otherwise just return empty because cannot make full block data
+                if (transactionsData.size != block.transactions.size) {
+                    log.warn("No data to fill the block")
+                    Mono.empty()
+                } else {
+                    joinWithTransactions(blockSplit.t1, blockSplit.t2, Flux.fromIterable(transactionsData).map { it.json!! })
+                            .reduce(ByteBuffer.allocate(block.json.size * 4), accumulate)
+                            .map { it.flip().array() }
+                            .map { json ->
+                                BlockContainer(block.height, block.hash, block.difficulty, block.timestamp,
+                                        true,
+                                        json,
+                                        null,
+                                        block.transactions
+                                )
+                            }
+                }
+            }
+        }
+    }
+
+    fun splitByTransactions(json: ByteArray): Tuple2<ByteArray, ByteArray> {
+        //TODO find a lib that implements Knuth-Morris-Pratt Pattern Matching Algorithm for byte arrays
+        // and reimplement without making a string copy from bytes
+
+        val s = String(json)
+        val fieldStart = s.indexOf("\"transactions\"")
+        val arrayStart = s.indexOf("[", fieldStart)
+        val arrayEnd = s.indexOf("]", arrayStart)
+
+        val head = s.substring(0, arrayStart + 1)
+        val tail = s.substring(arrayEnd, s.length)
+
+        return Tuples.of(head.toByteArray(), tail.toByteArray())
+    }
+
+    fun joinWithTransactions(head: ByteArray, tail: ByteArray, transactions: Flux<ByteArray>): Flux<ByteArray> {
+        val separator = Flux.range(0, Integer.MAX_VALUE)
+                .map { it != 0 }
+
+        val transactionsWithSeparator = transactions.zipWith(separator)
+                .flatMap {
+                    val tx = Flux.just(it.t1)
+                    if (it.t2) {
+                        Flux.concat(Flux.just(",".toByteArray()), tx)
+                    } else {
+                        tx
+                    }
+                }
+
+        return Flux.concat(
+                Flux.just(head),
+                transactionsWithSeparator,
+                Flux.just(tail)
+        )
+    }
+
 }
--- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt
+++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt
@@ -158,11 +158,15 @@ class NativeCallRouter(
            throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be block number")
        }
        val withTx = params[1].toString().toBoolean()
-        return if (withTx) {
-            log.warn("Block by number is not implemented")
-            null
+        var block = reader.blocksByHeightAsCont()
+                .read(number)
+        block = if (withTx) {
+            block.flatMap {
+                fullBlocksReader.read(it.hash)
+            }
        } else {
-            reader.blocksByHeightAsCont().read(number).map { it.json!! }
+            block
        }
+        return block.map { it.json!! }
    }
 }
--- a/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksMemCacheSpec.groovy
@@ -71,6 +71,7 @@ class BlocksMemCacheSpec extends Specification {

            cache.add(BlockContainer.from(block))
        }
+        cache.purge()

        def act1 = cache.read(BlockId.from(hash1)).block()
        def act2 = cache.read(BlockId.from(hash2)).block()
--- a/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/cache/CachesSpec.groovy
@@ -18,6 +18,7 @@ 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.TxContainer
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.infinitape.etherjar.domain.BlockHash
@@ -25,6 +26,7 @@ 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 reactor.core.publisher.Mono
 import spock.lang.Specification

 import java.time.Instant
@@ -145,14 +147,6 @@ class CachesSpec extends Specification {

    def "Cache txes of a requested block"() {
        setup:
-        TxMemCache txCache = Mock()
-        HeightCache heightCache = Mock()
-        BlocksMemCache blocksCache = Mock()
-        def caches = Caches.newBuilder()
-                .setTxByHash(txCache)
-                .setBlockByHeight(heightCache)
-                .setBlockByHash(blocksCache)
-                .build()

        def tx1 = new TransactionJson().with {
            hash = TransactionId.from(hash1)
@@ -166,20 +160,71 @@ class CachesSpec extends Specification {
            blockNumber = 100
            it
        }
+        BlockContainer block = new BlockJson().with { block ->
+            block.number = 100
+            block.hash = BlockHash.from(hash1)
+            block.totalDifficulty = BigInteger.ONE
+            block.transactions = [tx1, tx2]
+            block.timestamp = Instant.now()
+            BlockContainer.from(block)
+        }


-        def block = new BlockJson()
-        block.number = 100
-        block.hash = BlockHash.from(hash1)
-        block.totalDifficulty = BigInteger.ONE
-        block.transactions = [tx1, tx2]
-        block.timestamp = Instant.now()
-        block = BlockContainer.from(block)
+        TxMemCache txCache = Mock()
+        HeightCache heightCache = Mock()
+        BlocksMemCache blocksCache = Mock() {
+            _ * add(block)
+            _ * read(block.hash) >> Mono.just(block)
+        }
+        TxRedisCache txRedisCache = Mock()
+        def caches = Caches.newBuilder()
+                .setTxByHash(txCache)
+                .setBlockByHeight(heightCache)
+                .setBlockByHash(blocksCache)
+                .setTxByHash(txRedisCache)
+                .build()

        when:
        caches.cache(Caches.Tag.REQUESTED, block)
        then:
        1 * txCache.add(TxContainer.from(tx1))
        1 * txCache.add(TxContainer.from(tx2))
+        1 * txRedisCache.add(TxContainer.from(tx1), block) >> Mono.just(1).then()
+        1 * txRedisCache.add(TxContainer.from(tx2), block) >> Mono.just(1).then()
+    }
+
+    def "Cache tx with redis"() {
+        setup:
+
+        def tx1 = new TransactionJson().with {
+            hash = TransactionId.from(hash1)
+            blockHash = BlockHash.from(hash1)
+            blockNumber = 100
+            it
+        }
+        BlockContainer block = new BlockJson().with { block ->
+            block.number = 100
+            block.hash = BlockHash.from(hash1)
+            block.totalDifficulty = BigInteger.ONE
+            block.transactions = [tx1]
+            block.timestamp = Instant.now()
+            BlockContainer.from(block)
+        }
+        TxMemCache txCache = Mock()
+        HeightCache heightCache = Mock()
+        BlocksMemCache blocksCache = Mock()
+        TxRedisCache txRedisCache = Mock()
+        def caches = Caches.newBuilder()
+                .setTxByHash(txCache)
+                .setBlockByHeight(heightCache)
+                .setBlockByHash(blocksCache)
+                .setTxByHash(txRedisCache)
+                .build()
+
+        when:
+        caches.cache(Caches.Tag.REQUESTED, TxContainer.from(tx1))
+        then:
+        1 * blocksCache.read(block.hash) >> Mono.just(block)
+        1 * txRedisCache.add(TxContainer.from(tx1), block) >> Mono.just(1).then()
    }
 }
--- a/src/test/groovy/io/emeraldpay/dshackle/cache/HeightCacheSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/cache/HeightCacheSpec.groovy
@@ -72,6 +72,7 @@ class HeightCacheSpec extends Specification {
            block.timestamp = Instant.now()
            cache.add(BlockContainer.from(block))
        }
+        cache.purge()

        def act1 = cache.read(100).block()
        def act2 = cache.read(101).block()
--- a/src/test/groovy/io/emeraldpay/dshackle/cache/TxMemCacheSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/cache/TxMemCacheSpec.groovy
@@ -67,6 +67,7 @@ class TxMemCacheSpec extends Specification {
            tx.hash = TransactionId.from(hash)
            cache.add(TxContainer.from(tx))
        }
+        cache.purge()

        def act1 = cache.read(TxId.from(hash1)).block()
        def act2 = cache.read(TxId.from(hash2)).block()
--- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReaderSpec.groovy
+++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/EthereumFullBlocksReaderSpec.groovy
@@ -15,6 +15,8 @@
 */
 package io.emeraldpay.dshackle.upstream.ethereum

+import com.fasterxml.jackson.core.PrettyPrinter
+import com.fasterxml.jackson.core.util.DefaultPrettyPrinter
 import com.fasterxml.jackson.databind.ObjectMapper
 import io.emeraldpay.dshackle.Global
 import io.emeraldpay.dshackle.cache.BlocksMemCache
@@ -22,6 +24,8 @@ import io.emeraldpay.dshackle.cache.TxMemCache
 import io.emeraldpay.dshackle.data.BlockContainer
 import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.data.TxContainer
+import io.emeraldpay.dshackle.reader.Reader
+import io.emeraldpay.dshackle.test.ReaderMock
 import io.emeraldpay.dshackle.test.TestingCommons
 import io.infinitape.etherjar.domain.BlockHash
 import io.infinitape.etherjar.domain.TransactionId
@@ -200,7 +204,7 @@ class EthereumFullBlocksReaderSpec extends Specification {
        act.transactions.size() == 0
    }

-    def "Return nothing if no transactions"() {
+    def "Return nothing if some of tx are unavailable"() {
        setup:
        def txes = new TxMemCache()
        def blocks = new BlocksMemCache()
@@ -234,4 +238,140 @@ class EthereumFullBlocksReaderSpec extends Specification {
        then:
        act == null
    }
+
+    def "Doesn't change original cached json"() {
+        setup:
+        def txes = new TxMemCache()
+        def blocks = new BlocksMemCache()
+        def blockJson = '''
+        {
+            "number": "0x100001",
+            "hash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446",
+            "timestamp": "0x56cc7b8c",
+            "totalDifficulty": "0x6baba0399a0f2e73",
+            "extraField": "extraValue",
+            "transactions": [
+              "0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77",
+              "0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e"
+            ],
+            "extraField2": "extraValue2"
+        }
+        '''
+        blocks.add(BlockContainer.from(blockJson.bytes))
+
+        def tx1 = '''
+        {
+            "hash": "0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77",
+            "blockHash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446",
+            "blockNumber": "0x100001",
+            "extraField3": "extraValue3"
+        }
+        '''
+        def tx2 = '''
+        {
+            "hash": "0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e",
+            "blockHash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446",
+            "blockNumber": "0x100001",
+            "from": "0x2a65aca4d5fc5b5c859090a6c34d164135398226",
+            "extraField4": "extraValue4"
+        }
+        '''
+        txes.add(TxContainer.from(tx1.bytes))
+        txes.add(TxContainer.from(tx2.bytes))
+
+        def full = new EthereumFullBlocksReader(blocks, txes)
+
+        def blockJsonExpected = '''
+        {
+            "number": "0x100001",
+            "hash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446",
+            "timestamp": "0x56cc7b8c",
+            "totalDifficulty": "0x6baba0399a0f2e73",
+            "extraField": "extraValue",
+            "transactions": [
+        ''' + tx1 + ', ' + tx2 +
+                '''  ],
+            "extraField2": "extraValue2"
+        }
+        '''
+        blockJsonExpected = Global.objectMapper.readValue(blockJsonExpected, Map)
+
+        def prettyJson = Global.objectMapper.writer(new DefaultPrettyPrinter())
+        blockJsonExpected = prettyJson.writeValueAsString(blockJsonExpected)
+
+        when:
+        def act = full.read(BlockId.from("0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446")).block()
+        act = prettyJson.writeValueAsString(Global.objectMapper.readValue(act.json, Map))
+
+        then:
+        act.contains("extraValue")
+        act.contains("extraValue2")
+        act.contains("extraValue3")
+        act.contains("extraValue4")
+        act == blockJsonExpected
+    }
+
+    def "Split block with tx"() {
+        setup:
+        def blockJson = '''
+        {
+            "number": "0x100001",
+            "hash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446",
+            "timestamp": "0x56cc7b8c",
+            "totalDifficulty": "0x6baba0399a0f2e73",
+            "extraField": "extraValue",
+            "transactions": [
+              "0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77",
+              "0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e"
+            ],
+            "extraField2": "extraValue2"
+        }
+        '''
+        def reader = new EthereumFullBlocksReader(Stub(Reader), Stub(Reader))
+
+        when:
+        def act = reader.splitByTransactions(blockJson.bytes)
+        then:
+        new String(act.getT1()).endsWith('"transactions": [')
+        new String(act.getT2()).startsWith('],')
+        new String(act.getT2()).trim().endsWith('}')
+    }
+
+    def "Split block with tx if formatted with space"() {
+        setup:
+        def blockJson = '''
+        {
+            "number": "0x100001",
+            "hash": "0x18c68d9ba58772a4409d65d61891b25db03a105a7769ae08ef2cff697921b446",
+            "timestamp": "0x56cc7b8c",
+            "totalDifficulty": "0x6baba0399a0f2e73",
+            "extraField": "extraValue",
+            "transactions" : [
+              "0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77",
+              "0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e"
+            ] ,
+            "extraField2": "extraValue2"
+        }
+        '''
+        def reader = new EthereumFullBlocksReader(Stub(Reader), Stub(Reader))
+
+        when:
+        def act = reader.splitByTransactions(blockJson.bytes)
+        then:
+        new String(act.getT1()).endsWith('"transactions" : [')
+        new String(act.getT2()).startsWith('] ,')
+        new String(act.getT2()).trim().endsWith('}')
+    }
+
+    def "Split block with tx if formatted without space"() {
+        setup:
+        def blockJson = '{"extraField": "extraValue","transactions":["0x146b8f4b6300c73bb7476359b9f1c5ee3f686a86b2aa673552cf0f9de9a42e77","0xe589a39acea3091b584b650158d08b159aa07e97b8e8cddb8f81cb606e13382e"]}'
+        def reader = new EthereumFullBlocksReader(Stub(Reader), Stub(Reader))
+
+        when:
+        def act = reader.splitByTransactions(blockJson.bytes)
+        then:
+        new String(act.getT1()).endsWith('"transactions":[')
+        new String(act.getT2()) == ']}'
+    }
 }