Merge branch 'main' into jl/commit-kafka

Author: jakeloo

cmd/root.go

@@ -84,6 +84,7 @@ func init() {
 	rootCmd.PersistentFlags().Bool("storage-orchestrator-clickhouse-disableTLS", false, "Clickhouse disableTLS for orchestrator storage")
 	rootCmd.PersistentFlags().Bool("storage-orchestrator-clickhouse-enableParallelViewProcessing", false, "Clickhouse enableParallelViewProcessing for orchestrator storage")
 	rootCmd.PersistentFlags().Int("storage-orchestrator-clickhouse-maxQueryTime", 60, "Clickhouse max query time for orchestrator storage")
+	rootCmd.PersistentFlags().Int("storage-orchestrator-clickhouse-maxMemoryUsage", 1000000000, "Clickhouse max memory usage in bytes for orchestrator storage")
 	rootCmd.PersistentFlags().String("storage-orchestrator-postgres-host", "", "PostgreSQL host for orchestrator storage")
 	rootCmd.PersistentFlags().Int("storage-orchestrator-postgres-port", 5432, "PostgreSQL port for orchestrator storage")
 	rootCmd.PersistentFlags().String("storage-orchestrator-postgres-username", "", "PostgreSQL username for orchestrator storage")
@@ -105,6 +106,7 @@ func init() {
 	rootCmd.PersistentFlags().Bool("storage-main-clickhouse-disableTLS", false, "Clickhouse disableTLS for main storage")
 	rootCmd.PersistentFlags().Bool("storage-main-clickhouse-enableParallelViewProcessing", false, "Clickhouse enableParallelViewProcessing for main storage")
 	rootCmd.PersistentFlags().Int("storage-main-clickhouse-maxQueryTime", 60, "Clickhouse max query time for main storage")
+	rootCmd.PersistentFlags().Int("storage-main-clickhouse-maxMemoryUsage", 1000000000, "Clickhouse max memory usage in bytes for main storage")
 	rootCmd.PersistentFlags().String("storage-staging-clickhouse-username", "", "Clickhouse username for staging storage")
 	rootCmd.PersistentFlags().String("storage-staging-clickhouse-password", "", "Clickhouse password for staging storage")
 	rootCmd.PersistentFlags().Bool("storage-staging-clickhouse-asyncInsert", false, "Clickhouse async insert for staging storage")
@@ -114,6 +116,7 @@ func init() {
 	rootCmd.PersistentFlags().Bool("storage-staging-clickhouse-disableTLS", false, "Clickhouse disableTLS for staging storage")
 	rootCmd.PersistentFlags().Bool("storage-staging-clickhouse-enableParallelViewProcessing", false, "Clickhouse enableParallelViewProcessing for staging storage")
 	rootCmd.PersistentFlags().Int("storage-staging-clickhouse-maxQueryTime", 60, "Clickhouse max query time for staging storage")
+	rootCmd.PersistentFlags().Int("storage-staging-clickhouse-maxMemoryUsage", 1000000000, "Clickhouse max memory usage in bytes for staging storage")
 	rootCmd.PersistentFlags().String("storage-staging-postgres-host", "", "PostgreSQL host for staging storage")
 	rootCmd.PersistentFlags().Int("storage-staging-postgres-port", 5432, "PostgreSQL port for staging storage")
 	rootCmd.PersistentFlags().String("storage-staging-postgres-username", "", "PostgreSQL username for staging storage")
@@ -211,6 +214,7 @@ func init() {
 	viper.BindPFlag("storage.staging.clickhouse.disableTLS", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-disableTLS"))
 	viper.BindPFlag("storage.staging.clickhouse.enableParallelViewProcessing", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-enableParallelViewProcessing"))
 	viper.BindPFlag("storage.staging.clickhouse.maxQueryTime", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-maxQueryTime"))
+	viper.BindPFlag("storage.staging.clickhouse.maxMemoryUsage", rootCmd.PersistentFlags().Lookup("storage-staging-clickhouse-maxMemoryUsage"))
 	viper.BindPFlag("storage.main.clickhouse.database", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-database"))
 	viper.BindPFlag("storage.main.clickhouse.host", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-host"))
 	viper.BindPFlag("storage.main.clickhouse.port", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-port"))
@@ -223,6 +227,7 @@ func init() {
 	viper.BindPFlag("storage.main.clickhouse.disableTLS", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-disableTLS"))
 	viper.BindPFlag("storage.main.clickhouse.enableParallelViewProcessing", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-enableParallelViewProcessing"))
 	viper.BindPFlag("storage.main.clickhouse.maxQueryTime", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-maxQueryTime"))
+	viper.BindPFlag("storage.main.clickhouse.maxMemoryUsage", rootCmd.PersistentFlags().Lookup("storage-main-clickhouse-maxMemoryUsage"))
 	viper.BindPFlag("storage.orchestrator.clickhouse.database", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-database"))
 	viper.BindPFlag("storage.orchestrator.clickhouse.host", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-host"))
 	viper.BindPFlag("storage.orchestrator.clickhouse.port", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-port"))
@@ -235,6 +240,7 @@ func init() {
 	viper.BindPFlag("storage.orchestrator.clickhouse.disableTLS", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-disableTLS"))
 	viper.BindPFlag("storage.orchestrator.clickhouse.enableParallelViewProcessing", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-enableParallelViewProcessing"))
 	viper.BindPFlag("storage.orchestrator.clickhouse.maxQueryTime", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-maxQueryTime"))
+	viper.BindPFlag("storage.orchestrator.clickhouse.maxMemoryUsage", rootCmd.PersistentFlags().Lookup("storage-orchestrator-clickhouse-maxMemoryUsage"))
 	viper.BindPFlag("storage.orchestrator.postgres.host", rootCmd.PersistentFlags().Lookup("storage-orchestrator-postgres-host"))
 	viper.BindPFlag("storage.orchestrator.postgres.port", rootCmd.PersistentFlags().Lookup("storage-orchestrator-postgres-port"))
 	viper.BindPFlag("storage.orchestrator.postgres.username", rootCmd.PersistentFlags().Lookup("storage-orchestrator-postgres-username"))

configs/config.go

@@ -86,6 +86,7 @@ type ClickhouseConfig struct {
 	ChainBasedConfig             map[string]TableOverrideConfig `mapstructure:"chainBasedConfig"`
 	EnableParallelViewProcessing bool                           `mapstructure:"enableParallelViewProcessing"`
 	MaxQueryTime                 int                            `mapstructure:"maxQueryTime"`
+	MaxMemoryUsage               int                            `mapstructure:"maxMemoryUsage"`
 }
 
 type PostgresConfig struct {

internal/orchestrator/committer.go

@@ -96,6 +96,7 @@ func (c *Committer) Start(ctx context.Context) {
 		c.lastCommittedBlock.Store(latestCommittedBlockNumber.Uint64())
 	}
 
+	// Initialize publisher position - always use max(lastPublished, lastCommitted) to prevent double publishing
 	lastPublished, err := c.storage.StagingStorage.GetLastPublishedBlockNumber(chainID)
 	if err != nil {
 		// It's okay to fail silently here; it's only used for staging cleanup and will be
@@ -128,6 +129,54 @@ func (c *Committer) Start(ctx context.Context) {
 		c.lastPublishedBlock.Store(c.lastCommittedBlock.Load())
 	}
 
+	// Determine the correct publish position - always take the maximum to avoid going backwards
+	var targetPublishBlock *big.Int
+
+	if lastPublished == nil || lastPublished.Sign() == 0 {
+		// No previous publish position
+		if latestCommittedBlockNumber != nil && latestCommittedBlockNumber.Sign() > 0 {
+			// Start from committed position
+			targetPublishBlock = latestCommittedBlockNumber
+		} else if c.commitFromBlock.Sign() > 0 {
+			// Start from configured position minus 1 (since we publish from next block)
+			targetPublishBlock = new(big.Int).Sub(c.commitFromBlock, big.NewInt(1))
+		} else {
+			// Start from 0
+			targetPublishBlock = big.NewInt(0)
+		}
+
+		log.Info().
+			Str("target_publish_block", targetPublishBlock.String()).
+			Msg("No previous publish position, initializing publisher cursor")
+	} else {
+		// We have a previous position
+		targetPublishBlock = lastPublished
+	}
+
+	// Only update storage if we're changing the position
+	if lastPublished == nil || targetPublishBlock.Cmp(lastPublished) != 0 {
+		if err := c.storage.StagingStorage.SetLastPublishedBlockNumber(chainID, targetPublishBlock); err != nil {
+			log.Error().Err(err).Msg("Failed to update published block number in storage")
+			// If we can't update storage, use what was there originally to avoid issues
+			if lastPublished != nil {
+				targetPublishBlock = lastPublished
+			}
+		}
+	}
+
+	// Store in memory for quick acess
+	c.lastPublishedBlock.Store(targetPublishBlock.Uint64())
+
+	log.Info().
+		Str("publish_from", targetPublishBlock.String()).
+		Str("committed_at", func() string {
+			if latestCommittedBlockNumber != nil {
+				return latestCommittedBlockNumber.String()
+			}
+			return "0"
+		}()).
+		Msg("Publisher initialized")
+
 	c.cleanupProcessedStagingBlocks()
 
 	if config.Cfg.Publisher.Mode == "parallel" {
@@ -290,25 +339,27 @@ func (c *Committer) getBlockNumbersToCommit(ctx context.Context) ([]*big.Int, er
 }
 
 func (c *Committer) getBlockNumbersToPublish(ctx context.Context) ([]*big.Int, error) {
-	lastestPublishedBlockNumber, err := c.storage.StagingStorage.GetLastPublishedBlockNumber(c.rpc.GetChainID())
-	log.Debug().Msgf("Committer found this last published block number in staging storage: %s", lastestPublishedBlockNumber.String())
+	// Get the last published block from storage (which was already corrected in Start)
+	latestPublishedBlockNumber, err := c.storage.StagingStorage.GetLastPublishedBlockNumber(c.rpc.GetChainID())
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to get last published block number: %v", err)
 	}
 
-	if lastestPublishedBlockNumber.Sign() == 0 {
-		// If no blocks have been committed yet, start from the fromBlock specified in the config
-		lastestPublishedBlockNumber = new(big.Int).Sub(c.commitFromBlock, big.NewInt(1))
-	} else {
-		lastPublished := new(big.Int).SetUint64(c.lastPublishedBlock.Load())
-		if lastestPublishedBlockNumber.Cmp(lastPublished) < 0 {
-			log.Warn().Msgf("Max block in storage (%s) is less than last published block in memory (%s).", lastestPublishedBlockNumber.String(), lastPublished.String())
-			return []*big.Int{}, nil
-		}
+	// This should never happen after Start() has run, but handle it defensively
+	if latestPublishedBlockNumber == nil || latestPublishedBlockNumber.Sign() == 0 {
+		// Fall back to in-memory value which was set during Start
+		latestPublishedBlockNumber = new(big.Int).SetUint64(c.lastPublishedBlock.Load())
+		log.Warn().
+			Str("fallback_value", latestPublishedBlockNumber.String()).
+			Msg("Storage returned nil/0 for last published block, using in-memory value")
 	}
 
-	startBlock := new(big.Int).Add(lastestPublishedBlockNumber, big.NewInt(1))
-	endBlock, err := c.getBlockToCommitUntil(ctx, lastestPublishedBlockNumber)
+	log.Debug().
+		Str("last_published", latestPublishedBlockNumber.String()).
+		Msg("Determining blocks to publish")
+
+	startBlock := new(big.Int).Add(latestPublishedBlockNumber, big.NewInt(1))
+	endBlock, err := c.getBlockToCommitUntil(ctx, latestPublishedBlockNumber)
 	if err != nil {
 		return nil, fmt.Errorf("error getting block to commit until: %v", err)
 	}

internal/storage/clickhouse.go

@@ -670,8 +670,17 @@ func (c *ClickHouseConnector) addPostQueryClauses(query string, qf QueryFilter)
 	}
 
 	// Add settings at the very end
+	// Build settings string for ClickHouse query optimization
+	var settings []string
 	if c.cfg.MaxQueryTime > 0 {
-		query += fmt.Sprintf(" SETTINGS max_execution_time = %d", c.cfg.MaxQueryTime)
+		settings = append(settings, fmt.Sprintf("max_execution_time = %d", c.cfg.MaxQueryTime))
+	}
+	if c.cfg.MaxMemoryUsage > 0 {
+		settings = append(settings, fmt.Sprintf("max_memory_usage = %d", c.cfg.MaxMemoryUsage))
+	}
+
+	if len(settings) > 0 {
+		query += " SETTINGS " + strings.Join(settings, ", ")
 	}
 
 	return query