added protection for grpc head and grpc received blocks metric (#184)

src/main/kotlin/io/emeraldpay/dshackle/upstream/grpc/GrpcHead.kt

@@ -25,6 +25,8 @@ import io.emeraldpay.dshackle.upstream.DefaultUpstream
 import io.emeraldpay.dshackle.upstream.Lifecycle
 import io.emeraldpay.dshackle.upstream.UpstreamAvailability
 import io.emeraldpay.dshackle.upstream.forkchoice.ForkChoice
+import io.micrometer.core.instrument.Counter
+import io.micrometer.core.instrument.Metrics
 import org.reactivestreams.Publisher
 import reactor.core.Disposable
 import reactor.core.publisher.Flux
@@ -75,6 +77,10 @@ class GrpcHead(
                     .delayElements(Duration.ofSeconds(1))
             )
                 .flatMap { it.subscribeHead(chainRef) }
+                .doOnNext {
+                    headsCounter.increment()
+                }
+                .sample(Duration.ofMillis(1)) // protect from too many heads
                 .doFinally {
                     parent.setStatus(UpstreamAvailability.UNAVAILABLE)
                     log.warn("Head subscription finished: $it")
@@ -116,4 +122,9 @@ class GrpcHead(
         super.stop()
         headSubscription?.dispose()
     }
+
+    val headsCounter = Counter.builder("grpc_head_received")
+        .tag("upstream", id)
+        .tag("chain", chain.chainCode)
+        .register(Metrics.globalRegistry)
 }

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/EthereumGrpcUpstreamSpec.groovy

@@ -134,6 +134,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
 
             @Override
             void subscribeHead(Common.Chain request, StreamObserver<BlockchainOuterClass.ChainHead> responseObserver) {
+                new Thread({
                     responseObserver.onNext(
                             BlockchainOuterClass.ChainHead.newBuilder()
                                     .setBlockId(block1.hash.toHex().substring(2))
@@ -141,6 +142,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
                                     .setWeight(ByteString.copyFrom(block1.totalDifficulty.toByteArray()))
                                     .build()
                     )
+                    Thread.sleep(100)
                     responseObserver.onNext(
                             BlockchainOuterClass.ChainHead.newBuilder()
                                     .setBlockId(block2.hash.toHex().substring(2))
@@ -148,6 +150,7 @@ class EthereumGrpcUpstreamSpec extends Specification {
                                     .setWeight(ByteString.copyFrom(block2.totalDifficulty.toByteArray()))
                                     .build()
                     )
+                }).start()
             }
         })
         def upstream = new EthereumGrpcUpstream("test", hash, UpstreamsConfig.UpstreamRole.PRIMARY, Chain.ETHEREUM, client, new JsonRpcGrpcClient(client, Chain.ETHEREUM, metrics), null, ChainsConfig.ChainConfig.default())

src/test/groovy/io/emeraldpay/dshackle/upstream/grpc/GrpcHeadSpec.groovy

@@ -42,16 +42,20 @@ class GrpcHeadSpec extends Specification {
                     responseObserver.onError(new IllegalStateException("Unsupported chain"))
                     return
                 }
-                [10, 11, 12, 14].forEach { height ->
+                new Thread({ [10, 11, 12, 14].forEach { height ->
+                    println("next: $height")
                     responseObserver.onNext(
                             BlockchainOuterClass.ChainHead.newBuilder()
                                     .setChain(request.type)
+                                    .setBlockId(height.toString())
                                     .setHeight(height)
                                     .build()
                     )
-                    Thread.sleep(100)
+                    Thread.sleep(500)
                 }
                     responseObserver.onCompleted()
+                }).start()
+
             }
         })
         def convert = { BlockchainOuterClass.ChainHead head ->
@@ -88,6 +92,7 @@ class GrpcHeadSpec extends Specification {
                     responseObserver.onError(new IllegalStateException("Unsupported chain"))
                     return
                 }
+                new Thread({
                     phase++
                     if (phase == 1) {
                         responseObserver.onError(new RuntimeException("Phase 1 error"))
@@ -114,6 +119,7 @@ class GrpcHeadSpec extends Specification {
                         }
                         responseObserver.onCompleted()
                     }
+                }).start()
             }
         })
         def convert = { BlockchainOuterClass.ChainHead head ->