undo test codes

Author: nischitpra

cmd/committer.go

@@ -32,6 +32,6 @@ func RunCommitter(cmd *cobra.Command, args []string) {
 	committer.Init()
 	committer.InitReorg()
 
-	committer.RunReorgValidator()
-	// committer.CommitStreaming()
+	go committer.RunReorgValidator()
+	committer.CommitStreaming()
 }

internal/committer/reorg.go

@@ -33,7 +33,7 @@ func RunReorgValidator() {
 			continue
 		}
 
-		if endBlock-startBlock < 100 {
+		if endBlock == lastBlockCheck || endBlock-startBlock < 100 {
 			log.Debug().Int64("last_block_check", lastBlockCheck).Int64("start_block", startBlock).Int64("end_block", endBlock).Msg("Not enough new blocks to check. Sleeping for 1 minute.")
 			time.Sleep(1 * time.Minute)
 			continue

internal/libs/redis.go

@@ -12,7 +12,7 @@ import (
 
 var RedisClient *redis.Client
 
-const RedisReorgLastValidBlock = "reorg_last_valid_debug"
+const RedisReorgLastValidBlock = "reorg_last_valid"
 
 // InitRedis initializes the Redis client
 func InitRedis() {

internal/storage/kafka_publisher.go

@@ -232,10 +232,9 @@ func (p *KafkaPublisher) publishBlockData(blockData []*common.BlockData, isDelet
 		return nil
 	}
 
-	// test code, uncomment later
-	// if err := p.publishMessages(context.Background(), blockMessages); err != nil {
-	// 	return fmt.Errorf("failed to publish block messages: %v", err)
-	// }
+	if err := p.publishMessages(context.Background(), blockMessages); err != nil {
+		return fmt.Errorf("failed to publish block messages: %v", err)
+	}
 
 	log.Debug().Str("metric", "publish_duration").Msgf("Publisher.PublishBlockData duration: %f", time.Since(publishStart).Seconds())
 	return nil