fix: initialize publish cursor in parallel mode

Author: jakeloo

cmd/root.go

@@ -135,6 +135,7 @@ 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().Bool("publisher-blocks-enabled", false, "Toggle block publisher")
 	rootCmd.PersistentFlags().String("publisher-blocks-topicName", "", "Kafka topic name for blocks")
@@ -250,6 +251,7 @@ 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.blocks.enabled", rootCmd.PersistentFlags().Lookup("publisher-blocks-enabled"))
 	viper.BindPFlag("publisher.blocks.topicName", rootCmd.PersistentFlags().Lookup("publisher-blocks-topicName"))

configs/config.example.yml

@@ -190,9 +190,11 @@ api:
 publisher:
   # Whether the publisher is enabled
   enabled: true
+  # Publisher mode: "default" publishes after storage commit, "parallel" runs publishing alongside committing
+  mode: default
   # Kafka broker addresses (comma-separated)
   brokers: localhost:9092
-  
+
   # Block publishing configuration
   blocks:
     # Whether to publish block data

configs/config.go

@@ -172,6 +172,7 @@ 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"`

configs/test_config.yml

@@ -64,6 +64,7 @@ api:
 
 publisher:
   enabled: false
+  mode: default
 
 validation:
   mode: minimal 

internal/orchestrator/committer.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"math/big"
 	"sort"
+	"sync"
 	"time"
 
 	"github.com/rs/zerolog/log"
@@ -75,6 +76,34 @@ func NewCommitter(rpc rpc.IRPCClient, storage storage.IStorage, opts ...Committe
 	return committer
 }
 
+func (c *Committer) initializeParallelPublisher() {
+	chainID := c.rpc.GetChainID()
+	lastPublished, err := c.storage.StagingStorage.GetLastPublishedBlockNumber(chainID)
+	if err != nil {
+		log.Error().Err(err).Msg("failed to get last published block number")
+		return
+	}
+	mainMax, err := c.storage.MainStorage.GetMaxBlockNumber(chainID)
+	if err != nil {
+		log.Error().Err(err).Msg("failed to get max block number from main storage")
+		return
+	}
+	if lastPublished == nil || lastPublished.Sign() == 0 {
+		if mainMax != nil && mainMax.Sign() > 0 {
+			if err := c.storage.StagingStorage.SetLastPublishedBlockNumber(chainID, mainMax); err != nil {
+				log.Error().Err(err).Msg("failed to set last published block number")
+			}
+		}
+		return
+	}
+	if lastPublished.Cmp(mainMax) < 0 {
+		log.Warn().Msgf("Publish block number seek ahead from %s to %s", lastPublished.String(), mainMax.String())
+		if err := c.storage.StagingStorage.SetLastPublishedBlockNumber(chainID, mainMax); err != nil {
+			log.Error().Err(err).Msg("failed to set last published block number")
+		}
+	}
+}
+
 func (c *Committer) Start(ctx context.Context) {
 	interval := time.Duration(c.triggerIntervalMs) * time.Millisecond
 
@@ -83,11 +112,34 @@ func (c *Committer) Start(ctx context.Context) {
 	// Clean up staging data before starting the committer
 	c.cleanupStagingData()
 
+	if config.Cfg.Publisher.Mode == "parallel" {
+		c.initializeParallelPublisher()
+		var wg sync.WaitGroup
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			c.runCommitLoop(ctx, interval)
+		}()
+		go func() {
+			defer wg.Done()
+			c.runPublishLoop(ctx, interval)
+		}()
+		<-ctx.Done()
+		wg.Wait()
+		log.Info().Msg("Committer shutting down")
+		c.publisher.Close()
+		return
+	}
+
+	c.runCommitLoop(ctx, interval)
+	log.Info().Msg("Committer shutting down")
+	c.publisher.Close()
+}
+
+func (c *Committer) runCommitLoop(ctx context.Context, interval time.Duration) {
 	for {
 		select {
 		case <-ctx.Done():
-			log.Info().Msg("Committer shutting down")
-			c.publisher.Close()
 			return
 		case workMode := <-c.workModeChan:
 			if workMode != c.workMode && workMode != "" {
@@ -116,6 +168,24 @@ func (c *Committer) Start(ctx context.Context) {
 	}
 }
 
+func (c *Committer) runPublishLoop(ctx context.Context, interval time.Duration) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			time.Sleep(interval)
+			if c.workMode == "" {
+				log.Debug().Msg("Committer work mode not set, skipping publish")
+				continue
+			}
+			if err := c.publish(ctx); err != nil {
+				log.Error().Err(err).Msg("Error publishing blocks")
+			}
+		}
+	}
+}
+
 func (c *Committer) cleanupStagingData() {
 	// Get the last committed block number from main storage
 	latestCommittedBlockNumber, err := c.storage.MainStorage.GetMaxBlockNumber(c.rpc.GetChainID())
@@ -293,13 +363,88 @@ func (c *Committer) getSequentialBlockDataToCommit(ctx context.Context) ([]commo
 	return sequentialBlockData, nil
 }
 
+func (c *Committer) getSequentialBlockDataToPublish(ctx context.Context) ([]common.BlockData, error) {
+	chainID := c.rpc.GetChainID()
+	lastPublished, err := c.storage.StagingStorage.GetLastPublishedBlockNumber(chainID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get last published block number: %v", err)
+	}
+
+	startBlock := new(big.Int).Set(c.commitFromBlock)
+	if lastPublished != nil && lastPublished.Sign() > 0 {
+		startBlock = new(big.Int).Add(lastPublished, big.NewInt(1))
+	}
+
+	endBlock := new(big.Int).Add(startBlock, big.NewInt(int64(c.blocksPerCommit-1)))
+	blockCount := new(big.Int).Sub(endBlock, startBlock).Int64() + 1
+	blockNumbers := make([]*big.Int, blockCount)
+	for i := int64(0); i < blockCount; i++ {
+		blockNumbers[i] = new(big.Int).Add(startBlock, big.NewInt(i))
+	}
+
+	blocksData, err := c.storage.StagingStorage.GetStagingData(storage.QueryFilter{ChainId: chainID, BlockNumbers: blockNumbers})
+	if err != nil {
+		return nil, fmt.Errorf("error fetching blocks to publish: %v", err)
+	}
+	if len(blocksData) == 0 {
+		return nil, nil
+	}
+
+	sort.Slice(blocksData, func(i, j int) bool {
+		return blocksData[i].Block.Number.Cmp(blocksData[j].Block.Number) < 0
+	})
+	if blocksData[0].Block.Number.Cmp(startBlock) != 0 {
+		log.Debug().Msgf("First block to publish %s does not match expected %s", blocksData[0].Block.Number.String(), startBlock.String())
+		return nil, nil
+	}
+
+	sequential := []common.BlockData{blocksData[0]}
+	expected := new(big.Int).Add(blocksData[0].Block.Number, big.NewInt(1))
+	for i := 1; i < len(blocksData); i++ {
+		if blocksData[i].Block.Number.Cmp(blocksData[i-1].Block.Number) == 0 {
+			continue
+		}
+		if blocksData[i].Block.Number.Cmp(expected) != 0 {
+			break
+		}
+		sequential = append(sequential, blocksData[i])
+		expected.Add(expected, big.NewInt(1))
+	}
+
+	return sequential, nil
+}
+
+func (c *Committer) publish(ctx context.Context) error {
+	blockData, err := c.getSequentialBlockDataToPublish(ctx)
+	if err != nil {
+		return err
+	}
+	if len(blockData) == 0 {
+		return nil
+	}
+
+	if err := c.publisher.PublishBlockData(blockData); err != nil {
+		return err
+	}
+
+	chainID := c.rpc.GetChainID()
+	highest := blockData[len(blockData)-1].Block.Number
+	if err := c.storage.StagingStorage.SetLastPublishedBlockNumber(chainID, highest); err != nil {
+		return err
+	}
+	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
+		}
 	}
 	log.Debug().Msgf("Committing %d blocks", len(blockNumbers))
-
 	mainStorageStart := time.Now()
 	if err := c.storage.MainStorage.InsertBlockData(blockData); err != nil {
 		log.Error().Err(err).Msgf("Failed to commit blocks: %v", blockNumbers)
@@ -308,11 +453,13 @@ func (c *Committer) commit(ctx context.Context, blockData []common.BlockData) er
 	log.Debug().Str("metric", "main_storage_insert_duration").Msgf("MainStorage.InsertBlockData duration: %f", time.Since(mainStorageStart).Seconds())
 	metrics.MainStorageInsertDuration.Observe(time.Since(mainStorageStart).Seconds())
 
-	go func() {
-		if err := c.publisher.PublishBlockData(blockData); err != nil {
-			log.Error().Err(err).Msg("Failed to publish block data to kafka")
-		}
-	}()
+	if config.Cfg.Publisher.Mode == "default" {
+		go func() {
+			if err := c.publisher.PublishBlockData(blockData); err != nil {
+				log.Error().Err(err).Msg("Failed to publish block data to kafka")
+			}
+		}()
+	}
 
 	if c.workMode == WorkModeBackfill {
 		go func() {
@@ -325,13 +472,6 @@ func (c *Committer) commit(ctx context.Context, blockData []common.BlockData) er
 		}()
 	}
 
-	// Find highest block number from committed blocks
-	highestBlock := blockData[0].Block
-	for _, block := range blockData {
-		if block.Block.Number.Cmp(highestBlock.Number) > 0 {
-			highestBlock = block.Block
-		}
-	}
 	c.lastCommittedBlock = new(big.Int).Set(highestBlock.Number)
 
 	// Update metrics for successful commits