diff --git a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt index 93da6d38..ed34ecad 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/Global.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/Global.kt @@ -20,6 +20,8 @@ import com.fasterxml.jackson.databind.DeserializationFeature import com.fasterxml.jackson.databind.ObjectMapper import com.fasterxml.jackson.databind.module.SimpleModule import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse +import io.infinitape.etherjar.rpc.json.TransactionReceiptJson +import io.infinitape.etherjar.rpc.json.TransactionReceiptJsonDeserializer import java.text.SimpleDateFormat import java.util.* diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt index a584fed3..4c823029 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt @@ -27,61 +27,33 @@ import org.slf4j.LoggerFactory import reactor.core.publisher.Mono import java.math.BigInteger import java.time.Instant -import java.util.concurrent.TimeUnit -import kotlin.math.min /** * Cache blocks in Redis database */ class BlocksRedisCache( - private val redis: RedisReactiveCommands, - private val chain: Chain -) : Reader { + redis: RedisReactiveCommands, + chain: Chain +) : Reader, + OnBlockRedisCache(redis, chain, CachesProto.ValueContainer.ValueType.BLOCK) { companion object { private val log = LoggerFactory.getLogger(BlocksRedisCache::class.java) - private const val MAX_CACHE_TIME_MINUTES = 60L - - // doesn't make sense to cached in redis short living objects - private const val MIN_CACHE_TIME_SECONDS = 10 } - override fun read(key: BlockId): Mono { - return redis.get(key(key)) - .map { data -> - fromProto(data) - }.onErrorResume { - Mono.empty() - } - } - - fun toProto(value: BlockContainer): ByteArray { - if (value.full) { - throw IllegalArgumentException("Full Block is not supposed to be cached") - } - val meta = CachesProto.BlockMeta.newBuilder() - .setHash(ByteString.copyFrom(value.hash.value)) - .setHeight(value.height) - .setDifficulty(ByteString.copyFrom(value.difficulty.toByteArray())) - .setTimestamp(value.timestamp.toEpochMilli()) - - value.transactions.forEach { + override fun buildMeta(block: BlockContainer): CachesProto.BlockMeta.Builder { + val meta = super.buildMeta(block) + block.transactions.forEach { meta.addTxHashes(ByteString.copyFrom(it.value)) } - - return CachesProto.ValueContainer.newBuilder() - .setType(CachesProto.ValueContainer.ValueType.BLOCK) - .setValue(ByteString.copyFrom(value.json!!)) - .setBlockMeta(meta) - .build() - .toByteArray() + return meta } - fun fromProto(msg: ByteArray): BlockContainer { - val value = CachesProto.ValueContainer.parseFrom(msg) - if (value.type != CachesProto.ValueContainer.ValueType.BLOCK) { - throw IllegalArgumentException("Expect BLOCK value, receive ${value.type}") - } + override fun serializeValue(value: BlockContainer): ByteArray { + return value.json!! + } + + override fun deserializeValue(value: CachesProto.ValueContainer): BlockContainer { if (!value.hasBlockMeta()) { throw IllegalArgumentException("Container doesn't have Block Meta") } @@ -100,51 +72,14 @@ class BlocksRedisCache( ) } - fun evict(id: BlockId): Mono { - return Mono.just(id) - .flatMap { - redis.del(key(it)) - } - .then() - } - - /** - * Add to cache. - * Note that it returns Mono which must be subscribed to actually save - */ fun add(block: BlockContainer): Mono { if (block.timestamp == null || block.hash == null) { return Mono.empty() } - return Mono.just(block) - .flatMap { 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 - //still can be replaced in the blockchain - val age = Instant.now().epochSecond - block.timestamp!!.epochSecond - val ttl = min(age, TimeUnit.MINUTES.toSeconds(MAX_CACHE_TIME_MINUTES)) - if (ttl > MIN_CACHE_TIME_SECONDS) { - val key = key(block.hash) - val value = toProto(block) - redis.setex(key, ttl, value) - } else { - Mono.empty() - } - } - .doOnError { - log.warn("Failed to save Block to Redis: ${it.message}") - } - //if failed to cache, just continue without it - .onErrorResume { - Mono.empty() - } - .then() + if (block.full) { + return Mono.error(IllegalArgumentException("Full Block is not supposed to be cached")) + } + return super.add(block, block) } - /** - * Key in Redis - */ - fun key(hash: BlockId): String { - return "block:${chain.id}:${hash.toHex()}" - } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt index d2eb15d5..4f0bdd39 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt @@ -15,17 +15,16 @@ */ 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.data.TxId +import io.emeraldpay.dshackle.data.* import io.emeraldpay.dshackle.reader.CompoundReader +import io.emeraldpay.dshackle.reader.EmptyReader import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.Head import io.emeraldpay.dshackle.upstream.ethereum.EthereumFullBlocksReader import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionJson +import io.infinitape.etherjar.rpc.json.TransactionReceiptJson import org.slf4j.LoggerFactory import reactor.core.publisher.Flux import reactor.core.publisher.Mono @@ -35,7 +34,8 @@ open class Caches( private val blocksByHeight: HeightCache, private val memTxsByHash: TxMemCache, private val redisBlocksByHash: BlocksRedisCache?, - private val redisTxsByHash: TxRedisCache? + private val redisTxsByHash: TxRedisCache?, + private val redisReceipts: ReceiptRedisCache? ) { companion object { @@ -54,6 +54,7 @@ open class Caches( private val blocksByHash: Reader private val txsByHash: Reader + private val receiptByHash: Reader init { blocksByHash = if (redisBlocksByHash == null) { @@ -66,6 +67,12 @@ open class Caches( } else { CompoundReader(memTxsByHash, redisTxsByHash) } + receiptByHash = redisReceipts ?: EmptyReader() + } + + fun setHead(head: Head) { + redisTxsByHash?.head = head + redisReceipts?.head = head } /** @@ -79,6 +86,11 @@ open class Caches( } } + open fun cacheReceipt(tag: Tag, data: DefaultContainer) { + //TODO move subscription to the caller + redisReceipts?.add(data)?.subscribe() + } + fun cache(tag: Tag, tx: TxContainer) { //do not cache transactions that are not in a block yet if (tx.blockId == null) { @@ -166,6 +178,10 @@ open class Caches( return BlockByHeight(blocksByHeight, EthereumFullBlocksReader(blocksByHash, txsByHash)) } + fun getReceipts(): Reader { + return receiptByHash + } + enum class Tag { /** * Latest data produced by blockchain @@ -184,6 +200,7 @@ open class Caches( private var txsByHash: TxMemCache? = null private var redisBlocksByHash: BlocksRedisCache? = null private var redisTxsByHash: TxRedisCache? = null + private var redisReceiptCache: ReceiptRedisCache? = null fun setBlockByHash(cache: BlocksMemCache): Builder { blocksByHash = cache @@ -210,6 +227,11 @@ open class Caches( return this } + fun setReceipts(cache: ReceiptRedisCache): Builder { + redisReceiptCache = cache + return this + } + fun build(): Caches { if (blocksByHash == null) { blocksByHash = BlocksMemCache() @@ -220,7 +242,7 @@ open class Caches( if (txsByHash == null) { txsByHash = TxMemCache() } - return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, redisBlocksByHash, redisTxsByHash) + return Caches(blocksByHash!!, blocksByHeight!!, txsByHash!!, redisBlocksByHash, redisTxsByHash, redisReceiptCache) } } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt index 42e82349..71d32c60 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt @@ -15,7 +15,6 @@ */ package io.emeraldpay.dshackle.cache -import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.config.CacheConfig import io.emeraldpay.grpc.Chain import io.lettuce.core.RedisClient @@ -96,6 +95,7 @@ class CachesFactory( redis?.let { redis -> caches.setBlockByHash(BlocksRedisCache(redis.reactive(), chain)) caches.setTxByHash(TxRedisCache(redis.reactive(), chain)) + caches.setReceipts(ReceiptRedisCache(redis.reactive(), chain)) } return caches.build() } diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/OnBlockRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/OnBlockRedisCache.kt new file mode 100644 index 00000000..dd735d99 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/OnBlockRedisCache.kt @@ -0,0 +1,142 @@ +/** + * Copyright (c) 2020 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.google.protobuf.ByteString +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.proto.CachesProto +import io.emeraldpay.dshackle.proto.CachesProto.ValueContainer +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.grpc.Chain +import io.lettuce.core.api.reactive.RedisReactiveCommands +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono +import java.time.Instant +import java.util.concurrent.TimeUnit +import kotlin.math.min + +abstract class OnBlockRedisCache( + private val redis: RedisReactiveCommands, + private val chain: Chain, + private val valueType: ValueContainer.ValueType +) : Reader { + + companion object { + private val log = LoggerFactory.getLogger(OnBlockRedisCache::class.java) + + private const val MAX_CACHE_TIME_MINUTES = 60L + + // doesn't make sense to cached in redis short living objects + private const val MIN_CACHE_TIME_SECONDS = 3 + } + + private val prefix: String = when (valueType) { + ValueContainer.ValueType.BLOCK -> "block" + else -> throw IllegalStateException("No prefix for value type $valueType") + } + + fun toProto(block: BlockContainer, value: T): ValueContainer { + return ValueContainer.newBuilder() + .setType(valueType) + .setValue(ByteString.copyFrom(serializeValue(value))) + .setBlockMeta(buildMeta(block)) + .build() + } + + open fun buildMeta(block: BlockContainer): CachesProto.BlockMeta.Builder { + return CachesProto.BlockMeta.newBuilder() + .setHash(ByteString.copyFrom(block.hash.value)) + .setHeight(block.height) + .setDifficulty(ByteString.copyFrom(block.difficulty.toByteArray())) + .setTimestamp(block.timestamp.toEpochMilli()) + } + + abstract fun serializeValue(value: T): ByteArray + + fun fromProto(msg: ByteArray): T { + val value = ValueContainer.parseFrom(msg) + if (value.type != valueType) { + val error = "Expected $valueType value, received ${value.type}" + log.warn(error) + throw IllegalArgumentException(error) + } + return deserializeValue(value) + } + + abstract fun deserializeValue(value: ValueContainer): T + + /** + * Key in Redis + */ + fun key(hash: BlockId): String { + return "${prefix}:${chain.id}:${hash.toHex()}" + } + + /** + * Add to cache. + * Note that it returns Mono which must be subscribed to actually save + */ + open fun add(block: BlockContainer, value: T): Mono { + return Mono.just(block) + .flatMap { block -> + val ttl = cachingTime(block.timestamp!!) + if (ttl > MIN_CACHE_TIME_SECONDS) { + val key = key(block.hash) + val proto = toProto(block, value) + redis.setex(key, ttl, proto.toByteArray()) + } else { + Mono.empty() + } + } + .doOnError { + log.warn("Failed to save Block to Redis: ${it.message}") + } + //if failed to cache, just continue without it + .onErrorResume { + Mono.empty() + } + .then() + } + + /** + * Calculate time to cache the value + */ + fun cachingTime(blockTime: Instant): Long { + //default caching time is age of the block, i.e. block create hour ago + //keep for hour, but block created 10 seconds ago cache only for 10 seconds, because it + //still can be replaced in the blockchain + val age = Instant.now().epochSecond - blockTime.epochSecond + return min(age, TimeUnit.MINUTES.toSeconds(MAX_CACHE_TIME_MINUTES)) + } + + fun evict(id: BlockId): Mono { + return Mono.just(id) + .flatMap { + redis.del(key(it)) + } + .then() + } + + override fun read(key: BlockId): Mono { + return redis.get(key(key)) + .map { data -> + fromProto(data) + }.onErrorResume { + Mono.empty() + } + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/OnTxRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/OnTxRedisCache.kt new file mode 100644 index 00000000..1a40c656 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/OnTxRedisCache.kt @@ -0,0 +1,168 @@ +/** + * Copyright (c) 2020 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.google.protobuf.ByteString +import io.emeraldpay.dshackle.data.BlockContainer +import io.emeraldpay.dshackle.data.TxId +import io.emeraldpay.dshackle.proto.CachesProto +import io.emeraldpay.dshackle.reader.Reader +import io.emeraldpay.dshackle.upstream.Head +import io.emeraldpay.grpc.Chain +import io.lettuce.core.api.reactive.RedisReactiveCommands +import org.slf4j.LoggerFactory +import reactor.core.publisher.Mono +import java.time.Instant +import java.util.concurrent.TimeUnit +import kotlin.math.min + +abstract class OnTxRedisCache( + private val redis: RedisReactiveCommands, + private val chain: Chain, + private val valueType: CachesProto.ValueContainer.ValueType +) : Reader { + + companion object { + private val log = LoggerFactory.getLogger(OnTxRedisCache::class.java) + + // max caching time is 24 hours + const val MAX_CACHE_TIME_HOURS = 24L + const val MIN_CACHE_TIME_SECONDS = 30L + const val BLOCK_TIME_SECONDS = 10L + } + + private val prefix: String = when (valueType) { + CachesProto.ValueContainer.ValueType.TX -> "tx" + CachesProto.ValueContainer.ValueType.TX_RECEIPT -> "tx-receipt" + else -> throw IllegalStateException("No prefix for value type $valueType") + } + + var head: Head? = null + + /** + * Key in Redis + */ + fun key(hash: TxId): String { + return "${prefix}:${chain.id}:${hash.toHex()}" + } + + fun evict(block: BlockContainer): Mono { + return Mono.just(block) + .map { block -> + block.transactions.map { + key(it) + }.toTypedArray() + }.flatMap { keys -> + redis.del(*keys) + }.then() + } + + fun evict(id: TxId): Mono { + return Mono.just(id) + .flatMap { + redis.del(key(it)) + } + .then() + } + + fun toProto(id: TxId, value: T): ByteArray { + val meta = buildMeta(id, value) + + return CachesProto.ValueContainer.newBuilder() + .setType(valueType) + .setValue(ByteString.copyFrom(serializeValue(value))) + .setTxMeta(meta) + .build() + .toByteArray() + } + + open fun buildMeta(id: TxId, value: T): CachesProto.TxMeta.Builder { + return CachesProto.TxMeta.newBuilder() + .setHash(ByteString.copyFrom(id.value)) + } + + abstract fun serializeValue(value: T): ByteArray + + fun fromProto(msg: ByteArray): T { + val value = CachesProto.ValueContainer.parseFrom(msg) + if (value.type != valueType) { + val error = "Expected $valueType value, received ${value.type}" + log.warn(error) + throw IllegalArgumentException(error) + } + return deserializeValue(value) + } + + abstract fun deserializeValue(value: CachesProto.ValueContainer): T + + override fun read(key: TxId): Mono { + return redis.get(key(key)) + .map { data -> + fromProto(data) + }.onErrorResume { + Mono.empty() + } + } + + fun add(id: TxId, value: T, block: BlockContainer?, blockHeight: Long?): Mono { + return Mono.just(id) + .flatMap { + val key = key(it) + val encodedValue = toProto(it, value) + val ttl = if (block?.timestamp != null) { + cachingTime(block.timestamp) + } else { + cachingTime(blockHeight) + } + //store + redis.setex(key, ttl, encodedValue) + } + .doOnError { + log.warn("Failed to save TX to Redis: ${it.message}", it) + } + //if failed to cache, just continue without it + .onErrorResume { + Mono.empty() + } + .then() + } + + /** + * Calculate time to cache the value, based on block time + */ + fun cachingTime(blockTime: Instant): Long { + //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 - blockTime.epochSecond + return min(age, TimeUnit.HOURS.toSeconds(MAX_CACHE_TIME_HOURS)) + } + + /** + * Calculate time to cache the value, based on block height + */ + fun cachingTime(blockHeight: Long?): Long { + if (blockHeight == null) { + return MIN_CACHE_TIME_SECONDS + } + val headHeight = head?.getCurrentHeight() ?: return MIN_CACHE_TIME_SECONDS + val confirmations = headHeight - blockHeight + if (confirmations <= 0) { + return MIN_CACHE_TIME_SECONDS + } + return min(confirmations * BLOCK_TIME_SECONDS, TimeUnit.HOURS.toSeconds(MAX_CACHE_TIME_HOURS)) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptRedisCache.kt new file mode 100644 index 00000000..a6025b91 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptRedisCache.kt @@ -0,0 +1,42 @@ +/** + * Copyright (c) 2020 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 io.emeraldpay.dshackle.data.DefaultContainer +import io.emeraldpay.dshackle.data.TxId +import io.emeraldpay.dshackle.proto.CachesProto +import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.rpc.json.TransactionReceiptJson +import io.lettuce.core.api.reactive.RedisReactiveCommands +import reactor.core.publisher.Mono + +class ReceiptRedisCache( + redis: RedisReactiveCommands, + chain: Chain +) : OnTxRedisCache(redis, chain, CachesProto.ValueContainer.ValueType.TX_RECEIPT) { + + override fun deserializeValue(value: CachesProto.ValueContainer): ByteArray { + return value.value.toByteArray() + } + + override fun serializeValue(value: ByteArray): ByteArray { + return value + } + + fun add(json: DefaultContainer): Mono { + return super.add(json.txId!!, json.json!!, null, json.height) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt index 2d6f20fb..e5eaa14d 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt @@ -38,28 +38,15 @@ import kotlin.math.min class TxRedisCache( private val redis: RedisReactiveCommands, private val chain: Chain -) : Reader { +) : Reader, + OnTxRedisCache(redis, chain, CachesProto.ValueContainer.ValueType.TX) { 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: TxId): Mono { - return redis.get(key(key)) - .map { data -> - fromProto(data) - }.onErrorResume { - Mono.empty() - } - } - - fun toProto(value: TxContainer): ByteArray { - val meta = CachesProto.TxMeta.newBuilder() - .setHash(ByteString.copyFrom(value.hash.value)) - + override fun buildMeta(id: TxId, value: TxContainer): CachesProto.TxMeta.Builder { + val meta = super.buildMeta(id, value) value.height?.let { meta.setHeight(it) } @@ -67,20 +54,14 @@ class TxRedisCache( value.blockId?.value?.let { meta.setBlockHash(ByteString.copyFrom(it)) } - - return CachesProto.ValueContainer.newBuilder() - .setType(CachesProto.ValueContainer.ValueType.TX) - .setValue(ByteString.copyFrom(value.json!!)) - .setTxMeta(meta) - .build() - .toByteArray() + return meta } - fun fromProto(msg: ByteArray): TxContainer { - val value = CachesProto.ValueContainer.parseFrom(msg) - if (value.type != CachesProto.ValueContainer.ValueType.TX) { - throw IllegalArgumentException("Expect TX value, receive ${value.type}") - } + override fun serializeValue(value: TxContainer): ByteArray { + return value.json!! + } + + override fun deserializeValue(value: CachesProto.ValueContainer): TxContainer { if (!value.hasTxMeta()) { throw IllegalArgumentException("Container doesn't have Tx Meta") } @@ -93,55 +74,8 @@ class TxRedisCache( ) } - fun evict(block: BlockContainer): Mono { - return Mono.just(block) - .map { block -> - block.transactions.map { - key(it) - }.toTypedArray() - }.flatMap { keys -> - redis.del(*keys) - }.then() - } - - fun evict(id: TxId): Mono { - return Mono.just(id) - .flatMap { - redis.del(key(it)) - } - .then() - } - fun add(tx: TxContainer, block: BlockContainer): Mono { - if (tx.blockId == null || block.hash == null || tx.blockId != block.hash || block.timestamp == null) { - return Mono.empty() - } - return Mono.just(Tuples.of(tx, block)) - .flatMap { - val key = key(it.t1.hash) - val value = toProto(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)) - //store - redis.setex(key, ttl, value) - } - .doOnError { - log.warn("Failed to save TX to Redis: ${it.message}", it) - } - //if failed to cache, just continue without it - .onErrorResume { - Mono.empty() - } - .then() + return super.add(tx.hash, tx, block, tx.height) } - /** - * Key in Redis - */ - fun key(hash: TxId): String { - return "tx:${chain.id}:${hash.toHex()}" - } } \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/data/DefaultContainer.kt b/src/main/kotlin/io/emeraldpay/dshackle/data/DefaultContainer.kt new file mode 100644 index 00000000..ffc82389 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/data/DefaultContainer.kt @@ -0,0 +1,31 @@ +/** + * Copyright (c) 2020 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.data + +import org.slf4j.LoggerFactory + +class DefaultContainer( + val txId: TxId?, + val blockId: BlockId?, + val height: Long?, + json: ByteArray, + parsed: T +) : SourceContainer(json, parsed) { + + companion object { + private val log = LoggerFactory.getLogger(DefaultContainer::class.java) + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt b/src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt index 3519c3f7..b007c738 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt @@ -16,7 +16,6 @@ */ package io.emeraldpay.dshackle.data -import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.Global import io.infinitape.etherjar.rpc.json.TransactionJson diff --git a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt index ba2ca4b1..719745ae 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt @@ -144,6 +144,10 @@ open class NativeCall( .map { ctx.withPayload(it.value) } + .doOnNext { + ctx.upstream.postprocessor + .onReceive(ctx.payload.method, ctx.payload.params, it.payload) + } .onErrorMap { log.error("Failed to make a call", it) if (it is CallFailure) it diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt index 7b126d01..9bc0e389 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt @@ -44,7 +44,8 @@ import kotlin.concurrent.withLock abstract class Multistream( val chain: Chain, private val upstreams: MutableList, - val caches: Caches + val caches: Caches, + val postprocessor: RequestPostprocessor ) : Upstream, Lifecycle { companion object { @@ -189,6 +190,7 @@ abstract class Multistream( caches.cache(Caches.Tag.LATEST, it) } } + caches.setHead(head) } abstract fun updateHead(): Head diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt new file mode 100644 index 00000000..e8d7947b --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt @@ -0,0 +1,10 @@ +package io.emeraldpay.dshackle.upstream + +interface RequestPostprocessor { + + fun onReceive(method: String, params: List, json: ByteArray) + + class Empty : RequestPostprocessor { + override fun onReceive(method: String, params: List, json: ByteArray) {} + } +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt index f2652534..9de37b73 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt @@ -32,7 +32,7 @@ open class BitcoinMultistream( chain: Chain, val upstreams: MutableList, caches: Caches -) : Multistream(chain, upstreams as MutableList, caches), Lifecycle { +) : Multistream(chain, upstreams as MutableList, caches, RequestPostprocessor.Empty()), Lifecycle { companion object { private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt new file mode 100644 index 00000000..e824f808 --- /dev/null +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt @@ -0,0 +1,60 @@ +/** + * Copyright (c) 2020 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.upstream.ethereum + +import io.emeraldpay.dshackle.Global +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.RequestPostprocessor +import io.infinitape.etherjar.rpc.json.TransactionReceiptJson +import org.slf4j.LoggerFactory + +class CacheRequested( + private val caches: Caches +) : RequestPostprocessor { + + companion object { + private val log = LoggerFactory.getLogger(CacheRequested::class.java) + } + + override fun onReceive(method: String, params: List, json: ByteArray) { + try { + if (method == "eth_getTransactionReceipt") { + cacheTxReceipt(params, json) + } + } catch (e: Throwable) { + log.warn("Failed to cache result", e) + } + } + + fun cacheTxReceipt(params: List, json: ByteArray) { + if (params.size != 1) { + return + } + val parsed = Global.objectMapper.readValue(json, TransactionReceiptJson::class.java) + val value = DefaultContainer( + TxId.from(parsed.transactionHash), + BlockId.from(parsed.blockHash), + parsed.blockNumber, + json, + parsed + ) + caches.cacheReceipt(Caches.Tag.REQUESTED, value) + } + +} \ No newline at end of file diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt index 177e049a..7c56be78 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt @@ -31,7 +31,7 @@ open class EthereumMultistream( chain: Chain, val upstreams: MutableList, caches: Caches -) : Multistream(chain, upstreams as MutableList, caches) { +) : Multistream(chain, upstreams as MutableList, caches, CacheRequested(caches)) { companion object { private val log = LoggerFactory.getLogger(EthereumMultistream::class.java) diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt index 6b4a38c9..6f203a67 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt @@ -152,6 +152,10 @@ open class EthereumReader( ) } + fun receipts(): Reader { + return caches.getReceipts() + } + override fun isRunning(): Boolean { //TODO should be always running? return up.isRunning diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt index b1d5e484..813823a8 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt @@ -16,7 +16,6 @@ */ package io.emeraldpay.dshackle.upstream.ethereum -import com.fasterxml.jackson.databind.ObjectMapper import io.emeraldpay.dshackle.cache.Caches import io.emeraldpay.dshackle.cache.CachesEnabled import io.emeraldpay.dshackle.config.UpstreamsConfig diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt index c7637ebc..14fa8c3b 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt @@ -29,6 +29,13 @@ import org.slf4j.LoggerFactory import reactor.core.publisher.Mono import java.math.BigInteger +/** + * Reader for JSON RPC requests. Verifies if the method is allowed, transforms if necessary, and calls EthereumReader for data. + * It provides data only if it's available through the router (cached, head, etc). + * If data is not available locally then it returns `empty`; at this case the caller should call the remote node for actual data. + * + * @see EthereumReader + */ class NativeCallRouter( private val reader: EthereumReader, private val methods: CallMethods, @@ -100,6 +107,18 @@ class NativeCallRouter( method == "eth_getBlockByNumber" -> { getBlockByNumber(params) } + method == "eth_getTransactionReceipt" -> { + if (params.size != 1) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "Must provide 1 parameter") + } + val hash: TxId + try { + hash = TxId.from(params[0].toString()) + } catch (e: IllegalArgumentException) { + throw RpcException(RpcResponseError.CODE_INVALID_METHOD_PARAMS, "[0] must be transaction id") + } + reader.receipts().read(hash) + } else -> null } } diff --git a/src/main/proto/cache.proto b/src/main/proto/cache.proto index 9edbe6a3..b4c2f229 100644 --- a/src/main/proto/cache.proto +++ b/src/main/proto/cache.proto @@ -17,6 +17,7 @@ message ValueContainer { UNKNOWN = 0; BLOCK = 1; TX = 2; + TX_RECEIPT = 3; } enum Compression { diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy index 531e6cf6..1969735e 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy @@ -21,12 +21,10 @@ import io.emeraldpay.dshackle.data.BlockContainer import io.emeraldpay.dshackle.data.BlockId import io.emeraldpay.dshackle.data.TxId import io.emeraldpay.dshackle.test.IntegrationTestingCommons -import io.emeraldpay.dshackle.test.TestingCommons import io.emeraldpay.grpc.Chain import io.infinitape.etherjar.domain.BlockHash import io.infinitape.etherjar.rpc.json.BlockJson import io.infinitape.etherjar.rpc.json.TransactionRefJson -import io.lettuce.core.RedisClient import io.lettuce.core.api.StatefulRedisConnection import spock.lang.IgnoreIf import spock.lang.Specification @@ -69,8 +67,8 @@ class BlocksRedisCacheSpec extends Specification { ) when: - def enc = cache.toProto(cont) - def dec = cache.fromProto(enc) + def enc = cache.toProto(cont, cont) + def dec = cache.deserializeValue(enc) then: dec.height == 100 diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptRedisCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptRedisCacheSpec.groovy new file mode 100644 index 00000000..d1010e8c --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptRedisCacheSpec.groovy @@ -0,0 +1,96 @@ +package io.emeraldpay.dshackle.cache + +import com.fasterxml.jackson.databind.ObjectMapper +import io.emeraldpay.dshackle.Global +import io.emeraldpay.dshackle.data.BlockId +import io.emeraldpay.dshackle.data.DefaultContainer +import io.emeraldpay.dshackle.data.TxId +import io.emeraldpay.dshackle.test.IntegrationTestingCommons +import io.emeraldpay.grpc.Chain +import io.infinitape.etherjar.domain.BlockHash +import io.infinitape.etherjar.domain.TransactionId +import io.infinitape.etherjar.rpc.json.TransactionReceiptJson +import io.lettuce.core.api.StatefulRedisConnection +import spock.lang.IgnoreIf +import spock.lang.Specification + +@IgnoreIf({ IntegrationTestingCommons.isDisabled("redis") }) +class ReceiptRedisCacheSpec extends Specification { + + StatefulRedisConnection redis + ReceiptRedisCache cache + ObjectMapper objectMapper = Global.objectMapper + + String receiptJson = ''' + { + "blockHash": "0x2c3cfd4c7f2b58859371f5795eaf8524caa6e63145ac7e9df23c8d63aab891ae", + "blockNumber": "0x213b8a", + "contractAddress": null, + "cumulativeGasUsed": "0x5208", + "gasUsed": "0x5208", + "logs": [], + "transactionHash": "0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197", + "transactionIndex": "0x00" + } + ''' + TransactionReceiptJson receipt = new TransactionReceiptJson().tap { + transactionHash = TransactionId.from("0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197") + transactionIndex = 0 + blockHash = BlockHash.from("0x2c3cfd4c7f2b58859371f5795eaf8524caa6e63145ac7e9df23c8d63aab891ae") + blockNumber = 0x213b8a + cumulativeGasUsed = 0x5208 + gasUsed = 0x5208 + logs = [] + } + + def setup() { + redis = IntegrationTestingCommons.redisConnection() + redis.sync().flushdb() + cache = new ReceiptRedisCache( + redis.reactive(), Chain.ETHEREUM + ) + } + + def "Add and read"() { + setup: + + def container = new DefaultContainer( + TxId.from(receipt.transactionHash), + BlockId.from(receipt.blockHash), + receipt.blockNumber, + receiptJson.bytes, + receipt + ) + + when: + cache.add(container).block() + def act = cache.read(TxId.from(receipt.transactionHash)).block() + then: + act != null + objectMapper.readValue(act, TransactionReceiptJson) == receipt + } + + def "Evict value"() { + setup: + + def container = new DefaultContainer( + TxId.from(receipt.transactionHash), + BlockId.from(receipt.blockHash), + receipt.blockNumber, + receiptJson.bytes, + receipt + ) + + when: + cache.add(container).block() + def act = cache.read(TxId.from(receipt.transactionHash)).block() + then: + act != null + + when: + cache.evict(TxId.from(receipt.transactionHash)).subscribe() + act = cache.read(TxId.from(receipt.transactionHash)).block() + then: + act == null + } +} diff --git a/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy index cdd5eb5c..cf7396df 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy @@ -68,7 +68,7 @@ class TxRedisCacheSpec extends Specification { null ) when: - def enc = cache.toProto(cont) + def enc = cache.toProto(cont.hash, cont) def dec = cache.fromProto(enc) then: diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/CacheRequestedSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/CacheRequestedSpec.groovy new file mode 100644 index 00000000..9a97b5d5 --- /dev/null +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/CacheRequestedSpec.groovy @@ -0,0 +1,45 @@ +package io.emeraldpay.dshackle.upstream.ethereum + +import io.emeraldpay.dshackle.cache.Caches +import io.emeraldpay.dshackle.data.DefaultContainer +import spock.lang.Specification + +class CacheRequestedSpec extends Specification { + + def "Do nothing if unsupported method"() { + setup: + def caches = Mock(Caches) + CacheRequested instance = new CacheRequested(caches) + when: + instance.onReceive("eth_hashrate", [], '"0x38a"'.bytes) + then: + 0 * caches._(*_) + } + + def "Caches tx receipt"() { + setup: + def caches = Mock(Caches) + CacheRequested instance = new CacheRequested(caches) + def json = '''{ + "blockHash": "0x2c3cfd4c7f2b58859371f5795eaf8524caa6e63145ac7e9df23c8d63aab891ae", + "blockNumber": "0x213b8a", + "contractAddress": null, + "cumulativeGasUsed": "0x5208", + "gasUsed": "0x5208", + "logs": [], + "transactionHash": "0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197", + "transactionIndex": "0x00" + }'''.bytes + + when: + instance.onReceive("eth_getTransactionReceipt", ["0x5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197"], json) + + then: + 1 * caches.cacheReceipt(Caches.Tag.REQUESTED, { DefaultContainer it -> + it.height == 0x213b8a && + it.txId.toHex() == "5929b36be4586c57bd87dfb7ea6be3b985c1f527fa3d69d221604b424aeb4197" && + it.json == json + }) + } + +}