Fix generate identity hang and add more metrics (#666)

In a rare case when there are no logs in a batch of 1000, the batch
start point never moves and the entire indexer hangs. Fix it.

Also add metrics that were used to debug this.

Fixes #659 

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Enhanced connection monitoring with a new interceptor for managing
open connections in the API server.
- Refined logging across key operations, including block handling and
identity updates, with precise timestamp details.
- Integrated Prometheus-based metrics to track active API connections
for improved performance insights.

- **Bug Fixes**
- Corrected the logic for updating the `fromBlock` variable to ensure it
reflects the latest available block.

- **Documentation**
- Added detailed logging statements to improve observability of
operations.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

Author: mkysel

pkg/api/server.go

@@ -63,6 +63,11 @@ func NewAPIServer(
 		return nil, err
 	}
 
+	openConnectionsInterceptor, err := server.NewOpenConnectionsInterceptor()
+	if err != nil {
+		return nil, err
+	}
+
 	srvMetrics := grpcprom.NewServerMetrics(
 		grpcprom.WithServerHandlingTimeHistogram(
 			grpcprom.WithHistogramBuckets(
@@ -73,9 +78,13 @@ func NewAPIServer(
 
 	unary := []grpc.UnaryServerInterceptor{
 		srvMetrics.UnaryServerInterceptor(),
+		openConnectionsInterceptor.Unary(),
+		loggingInterceptor.Unary(),
 	}
 	stream := []grpc.StreamServerInterceptor{
 		srvMetrics.StreamServerInterceptor(),
+		openConnectionsInterceptor.Stream(),
+		loggingInterceptor.Stream(),
 	}
 
 	if jwtVerifier != nil {
@@ -95,9 +104,6 @@ func NewAPIServer(
 			PermitWithoutStream: true,
 			MinTime:             15 * time.Second,
 		}),
-		grpc.ChainUnaryInterceptor(loggingInterceptor.Unary()),
-		grpc.ChainStreamInterceptor(loggingInterceptor.Stream()),
-
 		// grpc.MaxRecvMsgSize(s.Config.Options.MaxMsgSize),
 	}
 

pkg/blockchain/rpcLogStreamer.go

@@ -168,6 +168,10 @@ func (r *RpcLogStreamer) watchContract(watcher ContractConfig) {
 			// reset self-termination timer
 			timer.Reset(watcher.maxDisconnectTime)
 
+			if nextBlock != nil {
+				fromBlock = *nextBlock
+			}
+
 			if len(logs) == 0 {
 				time.Sleep(NO_LOGS_SLEEP_TIME)
 				continue
@@ -177,14 +181,12 @@ func (r *RpcLogStreamer) watchContract(watcher ContractConfig) {
 				"Got logs",
 				zap.Int("numLogs", len(logs)),
 				zap.Uint64("fromBlock", fromBlock),
+				zap.Time("time", time.Now()),
 			)
 
 			for _, log := range logs {
 				watcher.EventChannel <- log
 			}
-			if nextBlock != nil {
-				fromBlock = *nextBlock
-			}
 		}
 	}
 }

pkg/indexer/storer/identityUpdate.go

@@ -58,7 +58,6 @@ func (s *IdentityUpdateStorer) StoreLog(
 	if err != nil {
 		return NewUnrecoverableLogStorageError(err)
 	}
-
 	err = db.RunInTx(
 		ctx,
 		s.db,
@@ -83,6 +82,7 @@ func (s *IdentityUpdateStorer) StoreLog(
 			s.logger.Info(
 				"Inserting identity update from contract",
 				zap.String("topic", messageTopic.String()),
+				zap.Time("time", time.Now()),
 			)
 
 			clientEnvelope, err := envelopes.NewClientEnvelopeFromBytes(msgSent.Update)

pkg/interceptors/server/openConnections.go

@@ -0,0 +1,44 @@
+package server
+
+import (
+	"context"
+	"github.com/xmtp/xmtpd/pkg/metrics"
+	"google.golang.org/grpc"
+)
+
+// OpenConnectionsInterceptor reports open connections for unary and stream RPCs.
+type OpenConnectionsInterceptor struct {
+}
+
+// NewOpenConnectionsInterceptor creates a new instance of OpenConnectionsInterceptor.
+func NewOpenConnectionsInterceptor() (*OpenConnectionsInterceptor, error) {
+	return &OpenConnectionsInterceptor{}, nil
+}
+
+// Unary intercepts unary RPC calls to log errors.
+func (i *OpenConnectionsInterceptor) Unary() grpc.UnaryServerInterceptor {
+	return func(
+		ctx context.Context,
+		req interface{},
+		info *grpc.UnaryServerInfo,
+		handler grpc.UnaryHandler,
+	) (interface{}, error) {
+		oc := metrics.NewApiOpenConnection("unary", info.FullMethod)
+		defer oc.Close()
+		return handler(ctx, req) // Call the actual RPC handler
+	}
+}
+
+// Stream intercepts stream RPC calls to log errors.
+func (i *OpenConnectionsInterceptor) Stream() grpc.StreamServerInterceptor {
+	return func(
+		srv interface{},
+		ss grpc.ServerStream,
+		info *grpc.StreamServerInfo,
+		handler grpc.StreamHandler,
+	) error {
+		oc := metrics.NewApiOpenConnection("stream", info.FullMethod)
+		defer oc.Close()
+		return handler(srv, ss) // Call the actual stream handler
+	}
+}

pkg/metrics/api.go

@@ -0,0 +1,33 @@
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var apiOpenConnections = prometheus.NewGaugeVec(
+	prometheus.GaugeOpts{
+		Name: "xmtp_api_open_connections_gauge",
+		Help: "Number of open API connections",
+	},
+	[]string{"style", "method"},
+)
+
+type ApiOpenConnection struct {
+	style  string
+	method string
+}
+
+func NewApiOpenConnection(style string, method string) *ApiOpenConnection {
+	oc := ApiOpenConnection{
+		style:  style,
+		method: method,
+	}
+
+	apiOpenConnections.With(prometheus.Labels{"style": oc.style, "method": oc.method}).Inc()
+
+	return &oc
+}
+
+func (oc *ApiOpenConnection) Close() {
+	apiOpenConnections.With(prometheus.Labels{"style": oc.style, "method": oc.method}).Dec()
+}

pkg/metrics/metrics.go

@@ -83,6 +83,7 @@ func registerCollectors(reg prometheus.Registerer) {
 		syncOutgoingSyncConnections,
 		syncFailedOutgoingSyncConnections,
 		syncFailedOutgoingSyncConnectionCounter,
+		apiOpenConnections,
 	}
 
 	for _, col := range cols {