do not close source in analyzers; rename vars; update comments

analyzer/consensus/consensus.go

@@ -90,8 +90,6 @@ func NewMain(cfg *config.BlockBasedAnalyzerConfig, genesisChainContext string, s
 
 // Start starts the main consensus analyzer.
 func (m *Main) Start(ctx context.Context) {
-	defer m.cleanup()
-
 	// Get block to be indexed.
 	var height int64
 
@@ -171,14 +169,6 @@ func (m *Main) Start(ctx context.Context) {
 			m.cfg.Range.From, m.cfg.Range.To))
 }
 
-func (m *Main) cleanup() {
-	if err := m.cfg.Source.Close(); err != nil {
-		m.logger.Error("failed to cleanly close consensus data source",
-			"err", err.Error(),
-		)
-	}
-}
-
 // Name returns the name of the Main.
 func (m *Main) Name() string {
 	return ConsensusAnalyzerName

analyzer/evmtokenbalances/evm_token_balances.go

@@ -247,8 +247,6 @@ func (m Main) processBatch(ctx context.Context) (int, error) {
 }
 
 func (m Main) Start(ctx context.Context) {
-	defer m.cleanup()
-
 	backoff, err := util.NewBackoff(
 		100*time.Millisecond,
 		// Cap the timeout at the expected round time. All runtimes currently have the same round time.
@@ -267,7 +265,6 @@ func (m Main) Start(ctx context.Context) {
 			// Process another batch of token balances.
 		case <-ctx.Done():
 			m.logger.Warn("shutting down evm_token_balances analyzer", "reason", ctx.Err())
-			m.cleanup()
 			return
 		}
 
@@ -289,12 +286,6 @@ func (m Main) Start(ctx context.Context) {
 	}
 }
 
-func (m *Main) cleanup() {
-	if err := m.cfg.Source.Close(); err != nil {
-		m.logger.Error("error closing data source", "err", err)
-	}
-}
-
 func (m Main) Name() string {
 	return EvmTokenBalancesAnalyzerPrefix + string(m.cfg.RuntimeName)
 }

analyzer/evmtokens/evm_tokens.go

@@ -189,8 +189,6 @@ func (m Main) processBatch(ctx context.Context) (int, error) {
 }
 
 func (m Main) Start(ctx context.Context) {
-	defer m.cleanup()
-
 	backoff, err := util.NewBackoff(
 		100*time.Millisecond,
 		// Cap the timeout at the expected round time. All runtimes currently have the same round time.
@@ -230,12 +228,6 @@ func (m Main) Start(ctx context.Context) {
 	}
 }
 
-func (m *Main) cleanup() {
-	if err := m.cfg.Source.Close(); err != nil {
-		m.logger.Error("error closing data source", "err", err)
-	}
-}
-
 func (m Main) Name() string {
 	return EvmTokensAnalyzerPrefix + string(m.cfg.RuntimeName)
 }

analyzer/runtime/runtime.go

@@ -40,7 +40,7 @@ var _ analyzer.Analyzer = (*Main)(nil)
 // NewRuntimeAnalyzer returns a new main analyzer for a runtime.
 func NewRuntimeAnalyzer(
 	runtime common.Runtime,
-	sdkPT *sdkConfig.ParaTime,
+	runtimeMetadata *sdkConfig.ParaTime,
 	cfg *config.BlockBasedAnalyzerConfig,
 	sourceClient *source.RuntimeClient,
 	target storage.TargetStorage,
@@ -52,7 +52,7 @@ func NewRuntimeAnalyzer(
 	}
 	ac := analyzer.RuntimeConfig{
 		RuntimeName: runtime,
-		ParaTime:    sdkPT,
+		ParaTime:    runtimeMetadata,
 		Range:       roundRange,
 		Source:      sourceClient,
 	}
@@ -66,8 +66,6 @@ func NewRuntimeAnalyzer(
 }
 
 func (m *Main) Start(ctx context.Context) {
-	defer m.cleanup()
-
 	if err := m.prework(ctx); err != nil {
 		m.logger.Error("error doing prework",
 			"err", err,
@@ -140,14 +138,6 @@ func (m *Main) Start(ctx context.Context) {
 			m.cfg.Range.From, m.cfg.Range.To))
 }
 
-func (m *Main) cleanup() {
-	if err := m.cfg.Source.Close(); err != nil {
-		m.logger.Error("failed to cleanly close consensus data source",
-			"err", err.Error(),
-		)
-	}
-}
-
 // Name returns the name of the Main.
 func (m *Main) Name() string {
 	return string(m.cfg.RuntimeName)

cmd/analyzer/analyzer.go

@@ -143,19 +143,27 @@ type Service struct {
 	logger  *log.Logger
 }
 
-// sourceFactory stores the sources used by all the analyzers in a Service.
+// sourceFactory stores singletons of the sources used by all the analyzers in a Service.
 // This enables re-use of node connections as well as graceful shutdown.
+// Note: NOT thread safe.
 type sourceFactory struct {
 	cfg config.SourceConfig
 
 	consensus *source.ConsensusClient
 	runtimes  map[common.Runtime]*source.RuntimeClient
 }
 
+func newSourceFactory(cfg config.SourceConfig) *sourceFactory {
+	return &sourceFactory{
+		cfg:      cfg,
+		runtimes: make(map[common.Runtime]*source.RuntimeClient),
+	}
+}
+
 func (s *sourceFactory) Close() error {
 	var firstErr error
 	if s.consensus != nil {
-		if err := s.consensus.Close(); err != nil && firstErr == nil {
+		if err := s.consensus.Close(); err != nil {
 			firstErr = err
 		}
 	}
@@ -218,9 +226,7 @@ func NewService(cfg *config.AnalysisConfig) (*Service, error) {
 	logger := cmdCommon.Logger().WithModule(moduleName)
 
 	// Initialize source storage.
-	sources := sourceFactory{
-		cfg: cfg.Source,
-	}
+	sources := newSourceFactory(cfg.Source)
 
 	// Initialize target storage.
 	dbClient, err := cmdCommon.NewClient(cfg.Storage, logger)
@@ -232,42 +238,42 @@ func NewService(cfg *config.AnalysisConfig) (*Service, error) {
 	analyzers := map[string]A{}
 	if cfg.Analyzers.Consensus != nil {
 		analyzers, err = addAnalyzer(analyzers, err, func() (A, error) {
-			chainContext := cfg.Source.History().CurrentRecord().ChainContext
+			genesisChainContext := cfg.Source.History().CurrentRecord().ChainContext
 			sourceClient, err1 := sources.Consensus(ctx)
 			if err1 != nil {
 				return nil, err1
 			}
-			return consensus.NewMain(cfg.Analyzers.Consensus, chainContext, sourceClient, dbClient, logger)
+			return consensus.NewMain(cfg.Analyzers.Consensus, genesisChainContext, sourceClient, dbClient, logger)
 		})
 	}
 	if cfg.Analyzers.Emerald != nil {
 		analyzers, err = addAnalyzer(analyzers, err, func() (A, error) {
-			sdkPT := cfg.Source.SDKParaTime(common.RuntimeEmerald)
+			runtimeMetadata := cfg.Source.SDKParaTime(common.RuntimeEmerald)
 			sourceClient, err1 := sources.Runtime(ctx, common.RuntimeEmerald)
 			if err1 != nil {
 				return nil, err1
 			}
-			return runtime.NewRuntimeAnalyzer(common.RuntimeEmerald, sdkPT, cfg.Analyzers.Emerald, sourceClient, dbClient, logger)
+			return runtime.NewRuntimeAnalyzer(common.RuntimeEmerald, runtimeMetadata, cfg.Analyzers.Emerald, sourceClient, dbClient, logger)
 		})
 	}
 	if cfg.Analyzers.Sapphire != nil {
 		analyzers, err = addAnalyzer(analyzers, err, func() (A, error) {
-			sdkPT := cfg.Source.SDKParaTime(common.RuntimeSapphire)
+			runtimeMetadata := cfg.Source.SDKParaTime(common.RuntimeSapphire)
 			sourceClient, err1 := sources.Runtime(ctx, common.RuntimeSapphire)
 			if err1 != nil {
 				return nil, err1
 			}
-			return runtime.NewRuntimeAnalyzer(common.RuntimeSapphire, sdkPT, cfg.Analyzers.Sapphire, sourceClient, dbClient, logger)
+			return runtime.NewRuntimeAnalyzer(common.RuntimeSapphire, runtimeMetadata, cfg.Analyzers.Sapphire, sourceClient, dbClient, logger)
 		})
 	}
 	if cfg.Analyzers.Cipher != nil {
 		analyzers, err = addAnalyzer(analyzers, err, func() (A, error) {
-			sdkPT := cfg.Source.SDKParaTime(common.RuntimeCipher)
+			runtimeMetadata := cfg.Source.SDKParaTime(common.RuntimeCipher)
 			sourceClient, err1 := sources.Runtime(ctx, common.RuntimeCipher)
 			if err1 != nil {
 				return nil, err1
 			}
-			return runtime.NewRuntimeAnalyzer(common.RuntimeCipher, sdkPT, cfg.Analyzers.Cipher, sourceClient, dbClient, logger)
+			return runtime.NewRuntimeAnalyzer(common.RuntimeCipher, runtimeMetadata, cfg.Analyzers.Cipher, sourceClient, dbClient, logger)
 		})
 	}
 	if cfg.Analyzers.EmeraldEvmTokens != nil {
@@ -325,7 +331,7 @@ func NewService(cfg *config.AnalysisConfig) (*Service, error) {
 	return &Service{
 		Analyzers: analyzers,
 
-		sources: &sources,
+		sources: sources,
 		target:  dbClient,
 		logger:  logger,
 	}, nil
@@ -379,8 +385,14 @@ func (a *Service) Start() {
 
 // cleanup cleans up resources used by the service.
 func (a *Service) cleanup() {
-	a.sources.Close()
+	if err := a.sources.Close(); err != nil {
+		a.logger.Error("failed to cleanly close data source",
+			"firstErr", err.Error(),
+		)
+	}
+	a.logger.Info("all source connections have closed cleanly")
 	a.target.Close()
+	a.logger.Info("indexer db connection closed cleanly")
 }
 
 // Register registers the process sub-command.