Fix: initialization

build/dipdup.testnet.yml

@@ -75,7 +75,7 @@ database:
   database: ${POSTGRES_DB:-dipdup}
 
 hasura:
-  url: http://hasura:8080
+  url: http://${HASURA_HOST:-hasura}:${HASURA_PORT:-8080}
   admin_secret: ${ADMIN_SECRET:-changeme}
   select_limit: 100
   allow_aggregation: false

cmd/mempool/main.go

@@ -19,6 +19,11 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+type startResult struct {
+	cancel  context.CancelFunc
+	indexer *Indexer
+}
+
 func main() {
 	log.SetFormatter(&log.TextFormatter{
 		FullTimestamp: true,
@@ -52,18 +57,21 @@ func main() {
 		prometheusService.Start()
 	}
 
+	started := make(chan struct{}, len(cfg.Mempool.Indexers))
 	indexerCancels := make(map[string]context.CancelFunc)
 	for network, mempool := range cfg.Mempool.Indexers {
 		for _, kind := range mempool.Filters.Kinds {
 			kinds[kind] = struct{}{}
 		}
 
 		go func(network string, mempool *config.Indexer) {
-			indexerCancel, err := startIndexer(ctx, network, cfg, mempool, prometheusService)
+			result, err := startIndexer(ctx, network, cfg, mempool, prometheusService)
 			if err != nil {
 				log.Error(err)
 			} else {
-				indexerCancels[network] = indexerCancel
+				indexers[network] = result.indexer
+				indexerCancels[network] = result.cancel
+				started <- struct{}{}
 				return
 			}
 
@@ -75,18 +83,22 @@ func main() {
 				case <-ctx.Done():
 					return
 				case <-ticker.C:
-					cancelFunc, err := startIndexer(ctx, network, cfg, mempool, prometheusService)
+					result, err := startIndexer(ctx, network, cfg, mempool, prometheusService)
 					if err != nil {
 						log.Error(err)
 					} else {
-						indexerCancels[network] = cancelFunc
+						indexers[network] = result.indexer
+						indexerCancels[network] = result.cancel
+						started <- struct{}{}
 						return
 					}
 				}
 			}
 		}(network, mempool)
 	}
 
+	<-started
+
 	views, err := createViews(ctx, cfg.Database)
 	if err != nil {
 		log.Error(err)
@@ -166,17 +178,21 @@ 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) {
+func startIndexer(ctx context.Context, network string, cfg config.Config, mempool *config.Indexer, prometheusService *prometheus.Service) (startResult, error) {
+	var result startResult
+
 	indexerCtx, cancel := context.WithCancel(ctx)
 	indexer, err := NewIndexer(indexerCtx, network, *mempool, cfg.Database, cfg.Mempool.Settings, prometheusService)
 	if err != nil {
 		cancel()
-		return nil, err
+		return result, err
 	}
+	result.indexer = indexer
 
 	if err := indexer.Start(indexerCtx); err != nil {
 		cancel()
-		return nil, err
+		return result, err
 	}
-	return cancel, nil
+	result.cancel = cancel
+	return result, nil
 }

docker-compose.yml

@@ -19,6 +19,8 @@ services:
     restart: always
     volumes:
       - db:/var/lib/postgres/data
+    ports:
+      - 127.0.0.1:5432:5432
     environment: 
       - POSTGRES_USER=dipdup
       - POSTGRES_DB=dipdup