Track web socket subscription_ids passed from grpc, retry eth_getLogs on errors/empty results (#465)

* retry getLogs when result is empty or error 

* print error logs with grpc subscription ids

* add logs on NativeSubscription's flux

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

@@ -108,15 +108,27 @@ class BlockchainRpc(
 
     override fun nativeSubscribe(request: Mono<BlockchainOuterClass.NativeSubscribeRequest>): Flux<BlockchainOuterClass.NativeSubscribeReplyItem> {
         var metrics: RequestMetrics? = null
-        return nativeSubscribe.nativeSubscribe(
+        return request
-            request
+            .doOnNext {
-                .doOnNext {
+                log.info("Starting subscription " + it.subscriptionId)
-                    metrics = chainMetrics.get(it.chain)
+                metrics = chainMetrics.get(it.chain)
-                    metrics!!.nativeSubscribeMetric.increment()
+                metrics!!.nativeSubscribeMetric.increment()
-                },
+            }.flatMapMany {
-        ).doOnNext {
+                    req ->
-            metrics?.nativeSubscribeRespMetric?.increment()
+                nativeSubscribe.nativeSubscribe(
-        }.doOnError { failMetric.increment() }
+                    Mono.just(req),
+                ).doOnNext {
+                    metrics?.nativeSubscribeRespMetric?.increment()
+                }.doOnComplete {
+                    log.info("Subscription ${req.subscriptionId} completed")
+                }.doOnCancel {
+                    log.info("Subscription ${req.subscriptionId} canceled")
+                }.doOnError {
+                        t ->
+                    log.info("Error ${t.message} in subscription ${req.subscriptionId}")
+                    failMetric.increment()
+                }
+            }
     }
 
     override fun subscribeHead(request: Mono<Common.Chain>): Flux<BlockchainOuterClass.ChainHead> {

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

@@ -46,20 +46,33 @@ open class NativeSubscribe(
     }
 
     private val objectMapper = Global.objectMapper
-
     fun nativeSubscribe(request: Mono<BlockchainOuterClass.NativeSubscribeRequest>): Flux<NativeSubscribeReplyItem> {
         return request
-            .flatMapMany(this@NativeSubscribe::start)
+            .flatMapMany {
-            .map(this@NativeSubscribe::convertToProto)
+                    it ->
-            .onErrorMap(this@NativeSubscribe::convertToStatus)
+                val subscriptionId = it.subscriptionId
+                Mono.just(it)
+                    .flatMapMany {
+                        start(it, subscriptionId)
+                    }
+                    .map(this@NativeSubscribe::convertToProto)
+                    .doOnCancel {
+                        log.warn("Subscription $subscriptionId cancelled")
+                    }.onErrorMap {
+                        convertToStatus(it, subscriptionId)
+                    }
+            }
     }
 
-    fun start(request: BlockchainOuterClass.NativeSubscribeRequest): Publisher<ResponseHolder> {
+    fun start(request: BlockchainOuterClass.NativeSubscribeRequest): Publisher<ResponseHolder> = start(request, "")
+
+    fun start(request: BlockchainOuterClass.NativeSubscribeRequest, subscriptionId: String): Publisher<ResponseHolder> {
         val chain = Chain.byId(request.chainValue)
 
         val multistream = getUpstream(chain)
 
         if (!multistream.getSubscriptionTopics().contains(request.method)) {
+            log.error("sub_id:" + subscriptionId + "subscribe ${request.method} is not supported for ${chain.chainCode}")
             return Mono.error(UnsupportedOperationException("subscribe ${request.method} is not supported for ${chain.chainCode}"))
         }
 
@@ -81,33 +94,38 @@ open class NativeSubscribe(
                     objectMapper.readValue(it.newInput(), List::class.java)
                 }
             }
-            subscribe(chain, method, params, matcher)
+            subscribe(chain, method, params, matcher, subscriptionId)
         }
         return publisher.map { ResponseHolder(it, nonce) }
     }
 
-    fun convertToStatus(t: Throwable) = when (t) {
+    fun convertToStatus(t: Throwable) = convertToStatus(t, "")
-        is SilentException.UnsupportedBlockchain -> StatusException(
+    fun convertToStatus(t: Throwable, subscriptionId: String = "") = when (t) {
-            Status.UNAVAILABLE.withDescription("BLOCKCHAIN UNAVAILABLE: ${t.blockchainId}"),
+        is SilentException.UnsupportedBlockchain -> {
-        )
+            log.error("sub_id:$subscriptionId BLOCKCHAIN UNAVAILABLE: ${t.blockchainId}")
+            StatusException(Status.UNAVAILABLE.withDescription("BLOCKCHAIN UNAVAILABLE: ${t.blockchainId}"))
+        }
 
-        is UnsupportedOperationException -> StatusException(
+        is UnsupportedOperationException -> {
-            Status.UNIMPLEMENTED.withDescription(t.message),
+            log.error("sub_id:$subscriptionId unimplemented error ${t.message}")
-        )
+            StatusException(Status.UNIMPLEMENTED.withDescription(t.message))
+        }
 
         else -> {
-            log.warn("Unhandled error", t)
+            log.warn("sub_id:$subscriptionId Unhandled error", t)
             StatusException(
                 Status.INTERNAL.withDescription(t.message),
             )
         }
     }
-
     open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher): Flux<out Any> =
+        subscribe(chain, method, params, matcher)
+
+    open fun subscribe(chain: Chain, method: String, params: Any?, matcher: Selector.Matcher, subscriptionId: String): Flux<out Any> =
         getUpstream(chain).getEgressSubscription()
             .subscribe(method, params, matcher)
             .doOnError {
-                log.error("Error during subscription to $method, chain $chain, params $params", it)
+                log.error("sub_id:$subscriptionId Error during subscription to $method, chain $chain, params $params", it)
             }
 
     private fun getUpstream(chain: Chain): Multistream =

src/main/kotlin/io/emeraldpay/dshackle/upstream/ethereum/subscribe/ProduceLogs.kt

@@ -21,6 +21,7 @@ import io.emeraldpay.dshackle.data.BlockId
 import io.emeraldpay.dshackle.reader.Reader
 import io.emeraldpay.dshackle.upstream.Multistream
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumCachingReader
+import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader
 import io.emeraldpay.dshackle.upstream.ethereum.EthereumDirectReader.Result
 import io.emeraldpay.dshackle.upstream.ethereum.hex.HexData
 import io.emeraldpay.dshackle.upstream.ethereum.json.TransactionLogJson
@@ -28,6 +29,7 @@ import io.emeraldpay.dshackle.upstream.ethereum.subscribe.json.LogMessage
 import org.slf4j.LoggerFactory
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
+import reactor.kotlin.core.publisher.onErrorResume
 import reactor.kotlin.core.publisher.switchIfEmpty
 import java.util.concurrent.TimeUnit
 
@@ -35,7 +37,7 @@ class ProduceLogs(
     private val logs: Reader<BlockId, Result<List<TransactionLogJson>>>,
     private val chain: Chain,
 ) {
-
+    private val MAX_RETRIES = 3
     companion object {
         private val log = LoggerFactory.getLogger(ProduceLogs::class.java)
     }
@@ -73,11 +75,26 @@ class ProduceLogs(
             .map { it.copy(removed = true) }
     }
 
-    fun produceAdded(update: ConnectBlockUpdates.Update): Flux<LogMessage> {
+    private fun produceAddedFallback(update: ConnectBlockUpdates.Update, retries: Int): Mono<EthereumDirectReader.Result<List<TransactionLogJson>>> {
         return logs.read(update.blockHash).switchIfEmpty {
-            log.warn("Cannot find receipt for block ${update.blockHash} for chain ${chain.chainName}")
+            if (retries > MAX_RETRIES) {
-            Mono.empty()
+                log.warn("Cannot find receipt for block ${update.blockHash} for chain ${chain.chainName} retries so far: $retries")
-        }.map {
+                Mono.empty()
+            } else {
+                produceAddedFallback(update, retries + 1)
+            }
+        }.onErrorResume { t ->
+            if (retries > MAX_RETRIES) {
+                log.error("Error ${t.message} produced ${update.blockHash} for chain ${chain.chainName} retries so far: $retries")
+                Mono.empty()
+            } else {
+                produceAddedFallback(update, retries + 1)
+            }
+        }
+    }
+
+    fun produceAdded(update: ConnectBlockUpdates.Update): Flux<LogMessage> {
+        return produceAddedFallback(update, 0).map {
             it.data
         }.flatMapMany {
             val messages = it.map { log ->