lower bound detector for receipts (#674)

* lower bound detector for receipts

* Add more non-retryable errors to dshackle proof detection

src/main/kotlin/io/emeraldpay/dshackle/rpc/ChainEventMapper.kt

@@ -150,6 +150,7 @@ class ChainEventMapper {
             LowerBoundType.PROOF -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_PROOF
             LowerBoundType.BLOB -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOB
             LowerBoundType.EPOCH -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_EPOCH
+            LowerBoundType.RECEIPTS -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_RECEIPTS
         }
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/rpc/StreamHead.kt

@@ -110,6 +110,7 @@ class StreamHead(
             LowerBoundType.PROOF -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_PROOF
             LowerBoundType.BLOB -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOB
             LowerBoundType.EPOCH -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_EPOCH
+            LowerBoundType.RECEIPTS -> BlockchainOuterClass.LowerBoundType.LOWER_BOUND_RECEIPTS
         }
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundProofDetector.kt

@@ -26,6 +26,8 @@ class EthereumLowerBoundProofDetector(
             "proofs are available only for the 'latest' block",
             "missing trie node",
             "cannot find EVM IAVL store",
+            "old data not available due to pruning",
+            "no historical RPC is available",
         )
     }
 

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundReceiptsDetector.kt

@@ -0,0 +1,76 @@
+package io.emeraldpay.dshackle.upstream.ethereum
+
+import io.emeraldpay.dshackle.Defaults
+import io.emeraldpay.dshackle.data.BlockContainer
+import io.emeraldpay.dshackle.upstream.ChainRequest
+import io.emeraldpay.dshackle.upstream.Upstream
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundData
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundDetector
+import io.emeraldpay.dshackle.upstream.lowerbound.LowerBoundType
+import io.emeraldpay.dshackle.upstream.lowerbound.detector.RecursiveLowerBound
+import io.emeraldpay.dshackle.upstream.lowerbound.toHex
+import io.emeraldpay.dshackle.upstream.rpcclient.ListParams
+import reactor.core.publisher.Flux
+
+class EthereumLowerBoundReceiptsDetector(
+    private val upstream: Upstream,
+) : LowerBoundDetector(upstream.getChain()) {
+
+    companion object {
+        const val MAX_OFFSET = 20
+        private const val NO_RECEIPTS_DATA = "No receipts data"
+
+        private val NO_RECEIPTS_ERRORS = setOf(
+            NO_RECEIPTS_DATA,
+            "block not found with number",
+            "requested epoch was a null round",
+            "missing trie node",
+            "old data not available due to pruning",
+        ).plus(EthereumLowerBoundBlockDetector.NO_BLOCK_ERRORS)
+    }
+
+    private val recursiveLowerBound = RecursiveLowerBound(upstream, LowerBoundType.RECEIPTS, NO_RECEIPTS_ERRORS, lowerBounds)
+
+    override fun period(): Long {
+        return 3
+    }
+
+    override fun internalDetectLowerBound(): Flux<LowerBoundData> {
+        return recursiveLowerBound.recursiveDetectLowerBoundWithOffset(MAX_OFFSET) { block ->
+            upstream.getIngressReader()
+                .read(
+                    ChainRequest("eth_getBlockByNumber", ListParams(block.toHex(), false)),
+                )
+                .timeout(Defaults.internalCallsTimeout)
+                .doOnNext {
+                    if (it.hasResult() && it.getResult().contentEquals("null".toByteArray())) {
+                        throw IllegalStateException(NO_RECEIPTS_DATA)
+                    }
+                }
+                .handle { it, sink ->
+                    val blockJson = BlockContainer.fromEthereumJson(it.getResult(), upstream.getId())
+                    if (blockJson.transactions.isEmpty()) {
+                        sink.error(IllegalStateException(NO_RECEIPTS_DATA))
+                        return@handle
+                    }
+                    sink.next(blockJson.transactions[0].toHexWithPrefix())
+                }
+                .flatMap { tx ->
+                    upstream.getIngressReader()
+                        .read(
+                            ChainRequest("eth_getTransactionReceipt", ListParams(tx)),
+                        )
+                        .timeout(Defaults.internalCallsTimeout)
+                        .doOnNext {
+                            if (it.hasResult() && it.getResult().contentEquals("null".toByteArray())) {
+                                throw IllegalStateException(NO_RECEIPTS_DATA)
+                            }
+                        }
+                }
+        }
+    }
+
+    override fun types(): Set<LowerBoundType> {
+        return setOf(LowerBoundType.RECEIPTS)
+    }
+}

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/EthereumLowerBoundService.kt

@@ -15,6 +15,7 @@ class EthereumLowerBoundService(
             EthereumLowerBoundBlockDetector(upstream),
             EthereumLowerBoundTxDetector(upstream),
             EthereumLowerBoundProofDetector(upstream),
+            EthereumLowerBoundReceiptsDetector(upstream),
         )
     }
 }

src/main/kotlin/io/emeraldpay/dshackle/upstream/lowerbound/LowerBoundData.kt

@@ -20,7 +20,7 @@ data class LowerBoundData(
 }
 
 enum class LowerBoundType {
-    UNKNOWN, STATE, SLOT, BLOCK, TX, LOGS, TRACE, PROOF, BLOB, EPOCH
+    UNKNOWN, STATE, SLOT, BLOCK, TX, LOGS, TRACE, PROOF, BLOB, EPOCH, RECEIPTS
 }
 
 fun BlockchainOuterClass.LowerBoundType.fromProtoType(): LowerBoundType {
@@ -36,5 +36,6 @@ fun BlockchainOuterClass.LowerBoundType.fromProtoType(): LowerBoundType {
         BlockchainOuterClass.LowerBoundType.LOWER_BOUND_PROOF -> LowerBoundType.PROOF
         BlockchainOuterClass.LowerBoundType.LOWER_BOUND_BLOB -> LowerBoundType.BLOB
         BlockchainOuterClass.LowerBoundType.LOWER_BOUND_EPOCH -> LowerBoundType.EPOCH
+        BlockchainOuterClass.LowerBoundType.LOWER_BOUND_RECEIPTS -> LowerBoundType.RECEIPTS
     }
 }