Fix: starting indexer process

Makefile

@@ -8,7 +8,7 @@ test:
 	go test ./...
 
 mempool:
-	cd cmd/mempool && go run . -c ../../build/dipdup.yml
+	cd cmd/mempool && go run . -c ../../build/dipdup.testnet.yml
 
 local:
 	docker-compose -f docker-compose.yml up -d --build

cmd/mempool/bakers.go

@@ -14,7 +14,7 @@ import (
 func (indexer *Indexer) setEndorsementBakers(ctx context.Context) {
 	defer indexer.wg.Done()
 
-	log.Info("Thread for finding endorsement baker started")
+	log.WithField("network", indexer.network).Info("Thread for finding endorsement baker started")
 
 	ticker := time.NewTicker(time.Second * 10)
 	defer ticker.Stop()

cmd/mempool/indexer.go

@@ -50,15 +50,15 @@ func NewIndexer(ctx context.Context, network string, indexerCfg config.Indexer,
 	}
 
 	rpc := node.NewNodeRPC(indexerCfg.DataSource.RPC[0])
-	constants, err := rpc.Constants()
+	constants, err := rpc.Constants(node.WithContext(ctx))
 	if err != nil {
 		return nil, err
 	}
 	if len(constants.TimeBetweenBlocks) == 0 {
 		return nil, errors.Errorf("Empty time_between_blocks in node response: %s", network)
 	}
 
-	head, err := rpc.Header("head")
+	head, err := rpc.Header("head", node.WithContext(ctx))
 	if err != nil {
 		return nil, err
 	}
@@ -74,7 +74,7 @@ func NewIndexer(ctx context.Context, network string, indexerCfg config.Indexer,
 
 	expiredAfter := settings.ExpiredAfter
 	if expiredAfter == 0 {
-		metadata, err := rpc.HeadMetadata("head")
+		metadata, err := rpc.HeadMetadata("head", node.WithContext(ctx))
 		if err != nil {
 			return nil, err
 		}

cmd/mempool/main.go

@@ -8,6 +8,7 @@ import (
 	"os/signal"
 	"strings"
 	"syscall"
+	"time"
 
 	"github.com/dipdup-net/go-lib/cmdline"
 	libCfg "github.com/dipdup-net/go-lib/config"
@@ -38,6 +39,8 @@ func main() {
 	signal.Notify(signals, os.Interrupt, syscall.SIGTERM, syscall.SIGINT)
 
 	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
 	indexers := make(map[string]*Indexer)
 
 	kinds := make(map[string]struct{})
@@ -49,23 +52,41 @@ func main() {
 		prometheusService.Start()
 	}
 
+	indexerCancels := make(map[string]context.CancelFunc)
 	for network, mempool := range cfg.Mempool.Indexers {
 		for _, kind := range mempool.Filters.Kinds {
 			kinds[kind] = struct{}{}
 		}
 
-		indexer, err := NewIndexer(ctx, network, *mempool, cfg.Database, cfg.Mempool.Settings, prometheusService)
-		if err != nil {
-			log.Error(err)
-			return
-		}
-		indexers[network] = indexer
-
-		if err := indexer.Start(ctx); err != nil {
-			log.Error(err)
-			return
-		}
+		go func(network string, mempool *config.Indexer) {
+			indexerCancel, err := startIndexer(ctx, network, cfg, mempool, prometheusService)
+			if err != nil {
+				log.Error(err)
+			} else {
+				indexerCancels[network] = indexerCancel
+				return
+			}
+
+			ticker := time.NewTicker(time.Minute)
+			defer ticker.Stop()
+
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case <-ticker.C:
+					cancelFunc, err := startIndexer(ctx, network, cfg, mempool, prometheusService)
+					if err != nil {
+						log.Error(err)
+					} else {
+						indexerCancels[network] = cancelFunc
+						return
+					}
+				}
+			}
+		}(network, mempool)
 	}
+
 	views, err := createViews(ctx, cfg.Database)
 	if err != nil {
 		log.Error(err)
@@ -85,9 +106,13 @@ func main() {
 	}
 
 	<-signals
-	cancel()
-
 	log.Warn("Trying carefully stopping....")
+
+	for network, indexerCancel := range indexerCancels {
+		log.Infof("stopping %s indexer...", network)
+		indexerCancel()
+	}
+
 	for _, indexer := range indexers {
 		if err := indexer.Close(); err != nil {
 			log.Error(err)
@@ -140,3 +165,18 @@ func createViews(ctx context.Context, database libCfg.Database) ([]string, error
 
 	return views, nil
 }
+
+func startIndexer(ctx context.Context, network string, cfg config.Config, mempool *config.Indexer, prometheusService *prometheus.Service) (context.CancelFunc, error) {
+	indexerCtx, cancel := context.WithCancel(ctx)
+	indexer, err := NewIndexer(indexerCtx, network, *mempool, cfg.Database, cfg.Mempool.Settings, prometheusService)
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+
+	if err := indexer.Start(indexerCtx); err != nil {
+		cancel()
+		return nil, err
+	}
+	return cancel, nil
+}