diff --git a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt index d6c1620a..5461072e 100644 --- a/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt +++ b/src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt @@ -45,7 +45,10 @@ class ProduceLogs( // need to keep history of recent messages in case they get removed. cannot rely on // any other cache or upstream because if when it gets removed it's unavailable in any other source private val oldMessages = CacheBuilder.newBuilder() - .expireAfterWrite(5, TimeUnit.HOURS) + // in general a block with its events can be replaced in ~90 seconds, but in case of a big network disturbance + // it can be much longer. here we keep events up to an hour + .expireAfterWrite(60, TimeUnit.MINUTES) + .maximumSize(90000) .build>() fun produce(block: Flux): Flux { @@ -74,8 +77,11 @@ class ProduceLogs( log.warn("Cannot find receipt for tx ${update.transactionId}") Mono.empty() } - .map { objectMapper.readValue(it, TransactionReceiptJson::class.java) } - .flatMapMany { receipt -> + .flatMapMany { jsonBytes -> + // receipt could be a null, like when the original block was replaced, etc. + // so just skip it as Flux.empty + val receipt = objectMapper.readValue(jsonBytes, TransactionReceiptJson::class.java) + ?: return@flatMapMany Flux.empty() try { val messages = receipt.logs .map { txlog -> diff --git a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy index 42bd85d0..c0d2ba67 100644 --- a/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy +++ b/src/test/groovy/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogsSpec.groovy @@ -55,6 +55,28 @@ class ProduceLogsSpec extends Specification { act.size() == 0 } + def "Produce added as nothing with null receipt"() { + setup: + String receipt = 'null' + + def receipts = Mock(Reader) { + 1 * it.read(TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af")) >> Mono.just(receipt.getBytes()) + } + def producer = new ProduceLogs(receipts) + def update = new ConnectBlockUpdates.Update( + BlockId.from("0x668b92d6b8c7db1350fd527fec4885ce5be2159b2b7daf6b126babdcbaa349da"), + 13412871, + ConnectBlockUpdates.UpdateType.NEW, + TxId.from("0x6c88df9d65ccc9351db65676c3581b29483e8dabb71c48ef7671c44b0d5568af") + ) + when: + def act = producer.produceAdded(update) + .collectList().block(Duration.ofSeconds(1)) + + then: + act.size() == 0 + } + def "Produce added with single log"() { setup: String receipt = '{\n' +