Remove publisher mode, default to parallel

Author: jakeloo

cmd/root.go

@@ -173,7 +173,6 @@ func init() {
 	rootCmd.PersistentFlags().Bool("api-contractApiRequest-disableCompression", false, "Disable compression for contract API request")
 	rootCmd.PersistentFlags().Int("api-contractApiRequest-timeout", 10, "Timeout in seconds for contract API request")
 	rootCmd.PersistentFlags().Bool("publisher-enabled", false, "Toggle publisher")
-	rootCmd.PersistentFlags().String("publisher-mode", "default", "Publisher mode: default or parallel")
 	rootCmd.PersistentFlags().String("publisher-brokers", "", "Kafka brokers")
 	rootCmd.PersistentFlags().String("publisher-username", "", "Kafka username for publisher")
 	rootCmd.PersistentFlags().String("publisher-password", "", "Kafka password for publisher")
@@ -368,7 +367,6 @@ func init() {
 	viper.BindPFlag("api.contractApiRequest.disableCompression", rootCmd.PersistentFlags().Lookup("api-contractApiRequest-disableCompression"))
 	viper.BindPFlag("api.contractApiRequest.timeout", rootCmd.PersistentFlags().Lookup("api-contractApiRequest-timeout"))
 	viper.BindPFlag("publisher.enabled", rootCmd.PersistentFlags().Lookup("publisher-enabled"))
-	viper.BindPFlag("publisher.mode", rootCmd.PersistentFlags().Lookup("publisher-mode"))
 	viper.BindPFlag("publisher.brokers", rootCmd.PersistentFlags().Lookup("publisher-brokers"))
 	viper.BindPFlag("publisher.username", rootCmd.PersistentFlags().Lookup("publisher-username"))
 	viper.BindPFlag("publisher.password", rootCmd.PersistentFlags().Lookup("publisher-password"))

configs/config.go

@@ -227,7 +227,6 @@ type EventPublisherConfig struct {
 
 type PublisherConfig struct {
 	Enabled      bool                       `mapstructure:"enabled"`
-	Mode         string                     `mapstructure:"mode"`
 	Brokers      string                     `mapstructure:"brokers"`
 	Username     string                     `mapstructure:"username"`
 	Password     string                     `mapstructure:"password"`

internal/orchestrator/committer.go

@@ -72,26 +72,22 @@ func NewCommitter(rpc rpc.IRPCClient, storage storage.IStorage, poller *Poller,
 func (c *Committer) Start(ctx context.Context) {
 	log.Debug().Msgf("Committer running")
 
-	if config.Cfg.Publisher.Mode == "parallel" {
-		var wg sync.WaitGroup
-		wg.Add(2)
+	var wg sync.WaitGroup
+	wg.Add(2)
 
-		go func() {
-			defer wg.Done()
-			c.runPublishLoop(ctx)
-		}()
+	go func() {
+		defer wg.Done()
+		c.runPublishLoop(ctx)
+	}()
 
-		go func() {
-			defer wg.Done()
-			c.runCommitLoop(ctx)
-		}()
+	go func() {
+		defer wg.Done()
+		c.runCommitLoop(ctx)
+	}()
 
-		<-ctx.Done()
+	<-ctx.Done()
 
-		wg.Wait()
-	} else {
-		c.runCommitLoop(ctx)
-	}
+	wg.Wait()
 
 	log.Info().Msg("Committer shutting down")
 	c.publisher.Close()
@@ -112,6 +108,7 @@ func (c *Committer) initCommittedAndPublishedBlockNumbers() error {
 	}
 	c.lastCommittedBlock.Store(latestCommittedBlockNumber.Uint64())
 
+	// Initialize published block number
 	lastPublished, err := c.storage.OrchestratorStorage.GetLastPublishedBlockNumber(c.rpc.GetChainID())
 	if err != nil {
 		return err
@@ -145,16 +142,7 @@ func (c *Committer) runCommitLoop(ctx context.Context) {
 				log.Info().Msgf("Committer reached configured toBlock %s, the last commit block is %d, stopping commits", c.commitToBlock.String(), c.lastCommittedBlock.Load())
 				return
 			}
-			blockDataToCommit, err := c.getSequentialBlockDataToCommit(ctx)
-			if err != nil {
-				log.Error().Err(err).Msg("Error getting block data to commit")
-				continue
-			}
-			if len(blockDataToCommit) == 0 {
-				log.Debug().Msg("No block data to commit")
-				continue
-			}
-			if err := c.commit(ctx, blockDataToCommit); err != nil {
+			if err := c.commit(ctx); err != nil {
 				log.Error().Err(err).Msg("Error committing blocks")
 			}
 			go c.cleanupProcessedStagingBlocks(ctx)
@@ -362,30 +350,16 @@ func (c *Committer) getSequentialBlockData(ctx context.Context, blockNumbers []*
 	return sequentialBlockData, nil
 }
 
-func (c *Committer) getSequentialBlockDataToCommit(ctx context.Context) ([]common.BlockData, error) {
-	blocksToCommit, err := c.getBlockNumbersToCommit(ctx)
-	if err != nil {
-		return nil, fmt.Errorf("error determining blocks to commit: %v", err)
-	}
-	if len(blocksToCommit) == 0 {
-		return nil, nil
-	}
-	return c.getSequentialBlockData(ctx, blocksToCommit)
-}
-
-func (c *Committer) getSequentialBlockDataToPublish(ctx context.Context) ([]common.BlockData, error) {
-	blocksToPublish, err := c.getBlockNumbersToPublish(ctx)
+func (c *Committer) publish(ctx context.Context) error {
+	blockNumbersToPublish, err := c.getBlockNumbersToPublish(ctx)
 	if err != nil {
-		return nil, fmt.Errorf("error determining blocks to publish: %v", err)
+		return fmt.Errorf("error determining blocks to publish: %v", err)
 	}
-	if len(blocksToPublish) == 0 {
-		return nil, nil
+	if len(blockNumbersToPublish) == 0 {
+		return nil
 	}
-	return c.getSequentialBlockData(ctx, blocksToPublish)
-}
 
-func (c *Committer) publish(ctx context.Context) error {
-	blockData, err := c.getSequentialBlockDataToPublish(ctx)
+	blockData, err := c.getSequentialBlockData(ctx, blockNumbersToPublish)
 	if err != nil {
 		return err
 	}
@@ -394,48 +368,51 @@ func (c *Committer) publish(ctx context.Context) error {
 	}
 
 	if err := c.publisher.PublishBlockData(blockData); err != nil {
+		log.Error().Err(err).Msgf("Failed to publish blocks: %v", blockNumbersToPublish)
 		return err
 	}
 
 	chainID := c.rpc.GetChainID()
 	highest := blockData[len(blockData)-1].Block.Number
 	if err := c.storage.OrchestratorStorage.SetLastPublishedBlockNumber(chainID, highest); err != nil {
+		log.Error().Err(err).Msgf("Failed to update last published block number to %s", highest.String())
 		return err
 	}
+
 	c.lastPublishedBlock.Store(highest.Uint64())
 	return nil
 }
 
-func (c *Committer) commit(ctx context.Context, blockData []common.BlockData) error {
-	blockNumbers := make([]*big.Int, len(blockData))
-	highestBlock := blockData[0].Block
-	for i, block := range blockData {
-		blockNumbers[i] = block.Block.Number
-		if block.Block.Number.Cmp(highestBlock.Number) > 0 {
-			highestBlock = block.Block
-		}
+func (c *Committer) commit(ctx context.Context) error {
+	blockNumbersToCommit, err := c.getBlockNumbersToCommit(ctx)
+	if err != nil {
+		return fmt.Errorf("error determining blocks to commit: %v", err)
+	}
+	if len(blockNumbersToCommit) == 0 {
+		return nil
+	}
+
+	blockData, err := c.getSequentialBlockData(ctx, blockNumbersToCommit)
+	if err != nil {
+		log.Error().Err(err).Msg("Error getting block data to commit")
+		return err
 	}
-	log.Debug().Msgf("Committing %d blocks from %s to %s", len(blockNumbers), blockNumbers[0].String(), blockNumbers[len(blockNumbers)-1].String())
+	if len(blockData) == 0 {
+		return nil
+	}
+
+	highestBlock := blockData[len(blockData)-1].Block
+
+	log.Debug().Msgf("Committing %d blocks from %s to %s", len(blockData), blockData[0].Block.Number.String(), highestBlock.Number.String())
 
 	mainStorageStart := time.Now()
 	if err := c.storage.MainStorage.InsertBlockData(blockData); err != nil {
-		log.Error().Err(err).Msgf("Failed to commit blocks: %v", blockNumbers)
+		log.Error().Err(err).Msgf("Failed to commit blocks: %v", blockNumbersToCommit)
 		return fmt.Errorf("error saving data to main storage: %v", err)
 	}
 	log.Debug().Str("metric", "main_storage_insert_duration").Msgf("MainStorage.InsertBlockData duration: %f", time.Since(mainStorageStart).Seconds())
 	metrics.MainStorageInsertDuration.Observe(time.Since(mainStorageStart).Seconds())
 
-	if config.Cfg.Publisher.Mode == "default" {
-		highest := highestBlock.Number.Uint64()
-		go func() {
-			if err := c.publisher.PublishBlockData(blockData); err != nil {
-				log.Error().Err(err).Msg("Failed to publish block data to kafka")
-				return
-			}
-			c.lastPublishedBlock.Store(highest)
-		}()
-	}
-
 	c.lastCommittedBlock.Store(highestBlock.Number.Uint64())
 
 	// Update metrics for successful commits