refactor: simplify publish loop

Author: jakeloo

internal/orchestrator/committer.go

@@ -174,31 +174,21 @@ func (c *Committer) runCommitLoop(ctx context.Context, interval time.Duration) {
 }
 
 func (c *Committer) runPublishLoop(ctx context.Context, interval time.Duration) {
-	chainID := c.rpc.GetChainID()
-	for {
-		select {
-		case <-ctx.Done():
-			return
-		default:
-			time.Sleep(interval)
-			if c.workMode == "" {
-				log.Debug().Msg("Committer work mode not set, skipping publish")
-				continue
-			}
-			lastPublished, err := c.storage.StagingStorage.GetLastPublishedBlockNumber(chainID)
-			if err != nil {
-				log.Error().Err(err).Msg("failed to get last published block number")
-				continue
-			}
-			lastCommitted := c.lastCommittedBlock.Load()
-			if lastPublished != nil && lastPublished.Uint64() >= lastCommitted {
-				continue
-			}
-			if err := c.publish(ctx); err != nil {
-				log.Error().Err(err).Msg("Error publishing blocks")
-			}
-		}
-	}
+        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) cleanupProcessedStagingBlocks() {

internal/orchestrator/committer_test.go

@@ -478,27 +478,25 @@ func TestRunPublishLoopPublishesWhenBehind(t *testing.T) {
 		{Block: common.Block{ChainId: chainID, Number: big.NewInt(102)}},
 	}
 
-	publishDone := make(chan struct{})
-	deleteDone := make(chan struct{})
-
-	mockRPC.EXPECT().GetChainID().Return(chainID)
-	mockStagingStorage.EXPECT().GetLastPublishedBlockNumber(chainID).Return(big.NewInt(100), nil)
-	mockRPC.EXPECT().GetChainID().Return(chainID)
-	mockStagingStorage.EXPECT().GetLastPublishedBlockNumber(chainID).Return(big.NewInt(100), nil)
-	mockStagingStorage.EXPECT().GetStagingData(mock.Anything).Return(blockData, nil)
-	mockRPC.EXPECT().GetChainID().Return(chainID)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	mockStagingStorage.EXPECT().SetLastPublishedBlockNumber(chainID, big.NewInt(102)).RunAndReturn(func(*big.Int, *big.Int) error {
-		close(publishDone)
-		cancel()
-		return nil
-	})
-	mockRPC.EXPECT().GetChainID().Return(chainID)
-	mockStagingStorage.EXPECT().DeleteOlderThan(chainID, big.NewInt(102)).RunAndReturn(func(*big.Int, *big.Int) error {
-		close(deleteDone)
-		return nil
-	})
+        publishDone := make(chan struct{})
+        deleteDone := make(chan struct{})
+
+        mockRPC.EXPECT().GetChainID().Return(chainID)
+        mockStagingStorage.EXPECT().GetLastPublishedBlockNumber(chainID).Return(big.NewInt(100), nil)
+        mockStagingStorage.EXPECT().GetStagingData(mock.Anything).Return(blockData, nil)
+        mockRPC.EXPECT().GetChainID().Return(chainID)
+
+        ctx, cancel := context.WithCancel(context.Background())
+        mockStagingStorage.EXPECT().SetLastPublishedBlockNumber(chainID, big.NewInt(102)).RunAndReturn(func(*big.Int, *big.Int) error {
+                close(publishDone)
+                cancel()
+                return nil
+        })
+        mockRPC.EXPECT().GetChainID().Return(chainID)
+        mockStagingStorage.EXPECT().DeleteOlderThan(chainID, big.NewInt(102)).RunAndReturn(func(*big.Int, *big.Int) error {
+                close(deleteDone)
+                return nil
+        })
 
 	go committer.runPublishLoop(ctx, time.Millisecond)
 
@@ -514,7 +512,7 @@ func TestRunPublishLoopPublishesWhenBehind(t *testing.T) {
 	}
 }
 
-func TestRunPublishLoopSkipsWhenAhead(t *testing.T) {
+func TestRunPublishLoopDoesNothingWhenNoNewBlocks(t *testing.T) {
 	defer func() { config.Cfg = config.Config{} }()
 	config.Cfg.Publisher.Mode = "parallel"
 	config.Cfg.Publisher.Enabled = false
@@ -532,20 +530,20 @@ func TestRunPublishLoopSkipsWhenAhead(t *testing.T) {
 	committer.workMode = WorkModeLive
 	committer.setLastCommittedBlock(big.NewInt(102))
 
-	chainID := big.NewInt(1)
+        chainID := big.NewInt(1)
 
-	mockRPC.EXPECT().GetChainID().Return(chainID)
-	mockStagingStorage.EXPECT().GetLastPublishedBlockNumber(chainID).Return(big.NewInt(105), nil)
+        mockRPC.EXPECT().GetChainID().Return(chainID)
+        mockStagingStorage.EXPECT().GetLastPublishedBlockNumber(chainID).Return(big.NewInt(105), nil)
+        mockStagingStorage.EXPECT().GetStagingData(mock.Anything).Return([]common.BlockData{}, nil)
 
-	ctx, cancel := context.WithCancel(context.Background())
-	go committer.runPublishLoop(ctx, time.Millisecond)
-	time.Sleep(2 * time.Millisecond)
-	cancel()
-	time.Sleep(10 * time.Millisecond)
+        ctx, cancel := context.WithCancel(context.Background())
+        go committer.runPublishLoop(ctx, time.Millisecond)
+        time.Sleep(2 * time.Millisecond)
+        cancel()
+        time.Sleep(10 * time.Millisecond)
 
-	mockStagingStorage.AssertNotCalled(t, "GetStagingData", mock.Anything)
-	mockStagingStorage.AssertNotCalled(t, "SetLastPublishedBlockNumber", mock.Anything, mock.Anything)
-	mockStagingStorage.AssertNotCalled(t, "DeleteOlderThan", mock.Anything, mock.Anything)
+        mockStagingStorage.AssertNotCalled(t, "SetLastPublishedBlockNumber", mock.Anything, mock.Anything)
+        mockStagingStorage.AssertNotCalled(t, "DeleteOlderThan", mock.Anything, mock.Anything)
 }
 
 func TestHandleGap(t *testing.T) {