StakeSquid/dshackle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #34 from emeraldpay/feat/cache-receipt

Browse Source 
solution: caching for eth_getTransactionReceipt
This commit is contained in:
Igor Artamonov
2020-07-12 23:39:19 -04:00
committed by GitHub
parent 3a93df2e32 b10e4800ff
commit 1cca38a7bc
24 changed files with 690 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/kotlin/io/emeraldpay/dshackle/Global.kt
View File

@@ -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.*

99
src/main/kotlin/io/emeraldpay/dshackle/cache/BlocksRedisCache.kt vendored
View File

@@ -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<String, ByteArray>,
private val chain: Chain
) : Reader<BlockId, BlockContainer> {
redis: RedisReactiveCommands<String, ByteArray>,
chain: Chain
) : Reader<BlockId, BlockContainer>,
OnBlockRedisCache<BlockContainer>(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<BlockContainer> {
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<Void> {
return Mono.just(id)
.flatMap {
redis.del(key(it))
}
.then()
}
/**
* Add to cache.
* Note that it returns Mono<Void> which must be subscribed to actually save
*/
fun add(block: BlockContainer): Mono<Void> {
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()}"
}
}

36
src/main/kotlin/io/emeraldpay/dshackle/cache/Caches.kt vendored
View File

@@ -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<BlockId, BlockContainer>
private val txsByHash: Reader<TxId, TxContainer>
private val receiptByHash: Reader<TxId, ByteArray>
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<TransactionReceiptJson>) {
//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<TxId, ByteArray> {
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)
}
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/cache/CachesFactory.kt vendored
View File

@@ -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()
}

142
src/main/kotlin/io/emeraldpay/dshackle/cache/OnBlockRedisCache.kt vendored Normal file
View File

@@ -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<T>(
private val redis: RedisReactiveCommands<String, ByteArray>,
private val chain: Chain,
private val valueType: ValueContainer.ValueType
) : Reader<BlockId, T> {
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<Void> which must be subscribed to actually save
*/
open fun add(block: BlockContainer, value: T): Mono<Void> {
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<Void> {
return Mono.just(id)
.flatMap {
redis.del(key(it))
}
.then()
}
override fun read(key: BlockId): Mono<T> {
return redis.get(key(key))
.map { data ->
fromProto(data)
}.onErrorResume {
Mono.empty()
}
}
}

168
src/main/kotlin/io/emeraldpay/dshackle/cache/OnTxRedisCache.kt vendored Normal file
View File

@@ -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<T>(
private val redis: RedisReactiveCommands<String, ByteArray>,
private val chain: Chain,
private val valueType: CachesProto.ValueContainer.ValueType
) : Reader<TxId, T> {
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<Void> {
return Mono.just(block)
.map { block ->
block.transactions.map {
key(it)
}.toTypedArray()
}.flatMap { keys ->
redis.del(*keys)
}.then()
}
fun evict(id: TxId): Mono<Void> {
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<T> {
return redis.get(key(key))
.map { data ->
fromProto(data)
}.onErrorResume {
Mono.empty()
}
}
fun add(id: TxId, value: T, block: BlockContainer?, blockHeight: Long?): Mono<Void> {
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))
}
}

42
src/main/kotlin/io/emeraldpay/dshackle/cache/ReceiptRedisCache.kt vendored Normal file
View File

@@ -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<String, ByteArray>,
chain: Chain
) : OnTxRedisCache<ByteArray>(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<TransactionReceiptJson>): Mono<Void> {
return super.add(json.txId!!, json.json!!, null, json.height)
}
}

88
src/main/kotlin/io/emeraldpay/dshackle/cache/TxRedisCache.kt vendored
View File

@@ -38,28 +38,15 @@ import kotlin.math.min
class TxRedisCache(
private val redis: RedisReactiveCommands<String, ByteArray>,
private val chain: Chain
) : Reader<TxId, TxContainer> {
) : Reader<TxId, TxContainer>,
OnTxRedisCache<TxContainer>(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<TxContainer> {
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<Void> {
return Mono.just(block)
.map { block ->
block.transactions.map {
key(it)
}.toTypedArray()
}.flatMap { keys ->
redis.del(*keys)
}.then()
}
fun evict(id: TxId): Mono<Void> {
return Mono.just(id)
.flatMap {
redis.del(key(it))
}
.then()
}
fun add(tx: TxContainer, block: BlockContainer): Mono<Void> {
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()}"
}
}

31
src/main/kotlin/io/emeraldpay/dshackle/data/DefaultContainer.kt Normal file
View File

@@ -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<T>(
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)
}
}

1
src/main/kotlin/io/emeraldpay/dshackle/data/TxContainer.kt
View File

@@ -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

4
src/main/kotlin/io/emeraldpay/dshackle/rpc/NativeCall.kt
View File

@@ -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

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/Multistream.kt
View File

@@ -44,7 +44,8 @@ import kotlin.concurrent.withLock
abstract class Multistream(
val chain: Chain,
private val upstreams: MutableList<Upstream>,
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

10
src/main/kotlin/io/emeraldpay/dshackle/upstream/RequestPostprocessor.kt Normal file
View File

@@ -0,0 +1,10 @@
package io.emeraldpay.dshackle.upstream
interface RequestPostprocessor {
fun onReceive(method: String, params: List<Any>, json: ByteArray)
class Empty : RequestPostprocessor {
override fun onReceive(method: String, params: List<Any>, json: ByteArray) {}
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/bitcoin/BitcoinMultistream.kt
View File

@@ -32,7 +32,7 @@ open class BitcoinMultistream(
chain: Chain,
val upstreams: MutableList<BitcoinUpstream>,
caches: Caches
) : Multistream(chain, upstreams as MutableList<Upstream>, caches), Lifecycle {
) : Multistream(chain, upstreams as MutableList<Upstream>, caches, RequestPostprocessor.Empty()), Lifecycle {
companion object {
private val log = LoggerFactory.getLogger(BitcoinMultistream::class.java)

60
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/CacheRequested.kt Normal file
View File

@@ -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<Any>, json: ByteArray) {
try {
if (method == "eth_getTransactionReceipt") {
cacheTxReceipt(params, json)
}
} catch (e: Throwable) {
log.warn("Failed to cache result", e)
}
}
fun cacheTxReceipt(params: List<Any>, json: ByteArray) {
if (params.size != 1) {
return
}
val parsed = Global.objectMapper.readValue(json, TransactionReceiptJson::class.java)
val value = DefaultContainer<TransactionReceiptJson>(
TxId.from(parsed.transactionHash),
BlockId.from(parsed.blockHash),
parsed.blockNumber,
json,
parsed
)
caches.cacheReceipt(Caches.Tag.REQUESTED, value)
}
}

2
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumMultistream.kt
View File

@@ -31,7 +31,7 @@ open class EthereumMultistream(
chain: Chain,
val upstreams: MutableList<EthereumUpstream>,
caches: Caches
) : Multistream(chain, upstreams as MutableList<Upstream>, caches) {
) : Multistream(chain, upstreams as MutableList<Upstream>, caches, CacheRequested(caches)) {
companion object {
private val log = LoggerFactory.getLogger(EthereumMultistream::class.java)

4
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumReader.kt
View File

@@ -152,6 +152,10 @@ open class EthereumReader(
)
}
fun receipts(): Reader<TxId, ByteArray> {
return caches.getReceipts()
}
override fun isRunning(): Boolean {
//TODO should be always running?
return up.isRunning

1
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumUpstream.kt
View File

@@ -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

19
src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/NativeCallRouter.kt
View File

@@ -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
}
}

1
src/main/proto/cache.proto
View File

@@ -17,6 +17,7 @@ message ValueContainer {
UNKNOWN = 0;
BLOCK = 1;
TX = 2;
TX_RECEIPT = 3;
}
enum Compression {

6
src/test/groovy/io/emeraldpay/dshackle/cache/BlocksRedisCacheSpec.groovy vendored
View File

@@ -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

96
src/test/groovy/io/emeraldpay/dshackle/cache/ReceiptRedisCacheSpec.groovy vendored Normal file
View File

@@ -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<String, byte[]> 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
}
}

2
src/test/groovy/io/emeraldpay/dshackle/cache/TxRedisCacheSpec.groovy vendored
View File

@@ -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:

45
src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/CacheRequestedSpec.groovy Normal file
View File

@@ -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
})
}
}