publish metrics only when actually publishing (#227)

### TL;DR

Moved metrics collection from the Committer to the Publisher for more accurate timing and consolidated metrics reporting.

### What changed?

- Removed metrics collection for publishing duration from the Committer's goroutine
- Moved the timing measurement for publishing to the Publisher itself
- Consolidated all metrics reporting (block counter, last published block, reorg counter) to happen in one place within the Publisher
- Fixed a bug where reorg metrics were only being recorded for old blocks, not new blocks

### How to test?

1. Run the application and verify that metrics are still being reported correctly
2. Check that the publish_duration metric now accurately reflects the actual time spent in the Publisher.PublishBlockData method
3. Verify that reorg metrics are properly recorded when a reorg occurs

### Why make this change?

The previous implementation had inaccurate timing measurements since the metrics were being recorded in the Committer's goroutine rather than in the Publisher itself. This change ensures that the publish_duration metric accurately reflects the actual time spent publishing data. Additionally, consolidating all metrics reporting in one place makes the code more maintainable and ensures consistent metrics collection.

Author: iuwqyir

internal/orchestrator/committer.go

@@ -253,13 +253,10 @@ 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())
 
-	publishStart := time.Now()
 	go func() {
 		if err := c.publisher.PublishBlockData(blockData); err != nil {
 			log.Error().Err(err).Msg("Failed to publish block data to kafka")
 		}
-		log.Debug().Str("metric", "publish_duration").Msgf("Publisher.PublishBlockData duration: %f", time.Since(publishStart).Seconds())
-		metrics.PublishDuration.Observe(time.Since(publishStart).Seconds())
 	}()
 
 	stagingDeleteStart := time.Now()

internal/publisher/publisher.go

@@ -97,13 +97,7 @@ func (p *Publisher) initialize() error {
 }
 
 func (p *Publisher) PublishBlockData(blockData []common.BlockData) error {
-	err := p.publishBlockData(blockData, false)
-	if err != nil {
-		return err
-	}
-	metrics.PublisherBlockCounter.Add(float64(len(blockData)))
-	metrics.LastPublishedBlock.Set(float64(blockData[len(blockData)-1].Block.Number.Int64()))
-	return nil
+	return p.publishBlockData(blockData, false)
 }
 
 func (p *Publisher) PublishReorg(oldData []common.BlockData, newData []common.BlockData) error {
@@ -114,8 +108,6 @@ func (p *Publisher) PublishReorg(oldData []common.BlockData, newData []common.Bl
 	if err := p.publishBlockData(newData, false); err != nil {
 		return fmt.Errorf("failed to publish new block data: %v", err)
 	}
-
-	metrics.PublisherReorgedBlockCounter.Add(float64(len(oldData)))
 	return nil
 }
 
@@ -168,6 +160,8 @@ func (p *Publisher) publishBlockData(blockData []common.BlockData, isReorg bool)
 		return nil
 	}
 
+	publishStart := time.Now()
+
 	// Prepare messages for blocks, events, transactions and traces
 	blockMessages := make([]*kgo.Record, len(blockData))
 	var eventMessages []*kgo.Record
@@ -251,6 +245,13 @@ func (p *Publisher) publishBlockData(blockData []common.BlockData, isReorg bool)
 		}
 	}
 
+	log.Debug().Str("metric", "publish_duration").Msgf("Publisher.PublishBlockData duration: %f", time.Since(publishStart).Seconds())
+	metrics.PublishDuration.Observe(time.Since(publishStart).Seconds())
+	metrics.PublisherBlockCounter.Add(float64(len(blockData)))
+	metrics.LastPublishedBlock.Set(float64(blockData[len(blockData)-1].Block.Number.Int64()))
+	if isReorg {
+		metrics.PublisherReorgedBlockCounter.Add(float64(len(blockData)))
+	}
 	return nil
 }