fix duplicate inserts

Author: iuwqyir

internal/orchestrator/committer.go

@@ -47,16 +47,15 @@ func NewCommitter(rpc rpc.IRPCClient, storage storage.IStorage) *Committer {
 
 func (c *Committer) Start(ctx context.Context) {
 	interval := time.Duration(c.triggerIntervalMs) * time.Millisecond
-	ticker := time.NewTicker(interval)
-	defer ticker.Stop()
 
 	log.Debug().Msgf("Committer running")
 	for {
 		select {
 		case <-ctx.Done():
 			log.Info().Msg("Committer shutting down")
 			return
-		case <-ticker.C:
+		default:
+			time.Sleep(interval)
 			blockDataToCommit, err := c.getSequentialBlockDataToCommit()
 			if err != nil {
 				log.Error().Err(err).Msg("Error getting block data to commit")

internal/orchestrator/committer_test.go

@@ -130,9 +130,11 @@ func TestCommit(t *testing.T) {
 	mockRPC := mocks.NewMockIRPCClient(t)
 	mockMainStorage := mocks.NewMockIMainStorage(t)
 	mockStagingStorage := mocks.NewMockIStagingStorage(t)
+	mockOrchestratorStorage := mocks.NewMockIOrchestratorStorage(t)
 	mockStorage := storage.IStorage{
-		MainStorage:    mockMainStorage,
-		StagingStorage: mockStagingStorage,
+		MainStorage:         mockMainStorage,
+		StagingStorage:      mockStagingStorage,
+		OrchestratorStorage: mockOrchestratorStorage,
 	}
 	committer := NewCommitter(mockRPC, mockStorage)
 
@@ -186,10 +188,12 @@ func TestStartCommitter(t *testing.T) {
 	mockRPC := mocks.NewMockIRPCClient(t)
 	mockMainStorage := mocks.NewMockIMainStorage(t)
 	mockStagingStorage := mocks.NewMockIStagingStorage(t)
+	mockOrchestratorStorage := mocks.NewMockIOrchestratorStorage(t)
 
 	mockStorage := storage.IStorage{
-		MainStorage:    mockMainStorage,
-		StagingStorage: mockStagingStorage,
+		MainStorage:         mockMainStorage,
+		StagingStorage:      mockStagingStorage,
+		OrchestratorStorage: mockOrchestratorStorage,
 	}
 
 	committer := NewCommitter(mockRPC, mockStorage)
@@ -218,9 +222,11 @@ func TestCommitterRespectsSIGTERM(t *testing.T) {
 	mockRPC := mocks.NewMockIRPCClient(t)
 	mockMainStorage := mocks.NewMockIMainStorage(t)
 	mockStagingStorage := mocks.NewMockIStagingStorage(t)
+	mockOrchestratorStorage := mocks.NewMockIOrchestratorStorage(t)
 	mockStorage := storage.IStorage{
-		MainStorage:    mockMainStorage,
-		StagingStorage: mockStagingStorage,
+		MainStorage:         mockMainStorage,
+		StagingStorage:      mockStagingStorage,
+		OrchestratorStorage: mockOrchestratorStorage,
 	}
 
 	committer := NewCommitter(mockRPC, mockStorage)

internal/storage/clickhouse.go

@@ -1082,18 +1082,59 @@ func (c *ClickHouseConnector) deleteTraces(chainId *big.Int, blockNumbers []*big
 	})
 }
 
+func (c *ClickHouseConnector) getBlockExistenceMap(data *[]common.BlockData) (map[string]bool, error) {
+	if len(*data) == 0 {
+		return nil, nil
+	}
+	blockNumbers := make([]*big.Int, 0, len(*data))
+	for _, blockData := range *data {
+		blockNumbers = append(blockNumbers, blockData.Block.Number)
+	}
+	chainId := (*data)[0].Block.ChainId
+	query := fmt.Sprintf("SELECT block_number FROM %s.%s WHERE chain_id = ? AND block_number IN (?) GROUP BY block_number HAVING sum(sign) > 0", c.cfg.Database, c.getTableName(chainId, "blocks"))
+	rows, err := c.conn.Query(context.Background(), query, chainId, blockNumbers)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	blockNumbersMap := make(map[string]bool)
+	for _, blockNumber := range blockNumbers {
+		blockNumbersMap[blockNumber.String()] = false
+	}
+	for rows.Next() {
+		var blockNumber string
+		err := rows.Scan(&blockNumber)
+		if err != nil {
+			return nil, err
+		}
+		blockNumbersMap[blockNumber] = true
+	}
+	return blockNumbersMap, nil
+}
+
 // TODO make this atomic
 func (c *ClickHouseConnector) InsertBlockData(data *[]common.BlockData) error {
+	if len(*data) == 0 {
+		return nil
+	}
 	blocks := make([]common.Block, 0, len(*data))
 	logs := make([]common.Log, 0)
 	transactions := make([]common.Transaction, 0)
 	traces := make([]common.Trace, 0)
 
+	blockExistenceMap, err := c.getBlockExistenceMap(data)
+	if err != nil {
+		return err
+	}
+
 	for _, blockData := range *data {
-		blocks = append(blocks, blockData.Block)
-		logs = append(logs, blockData.Logs...)
-		transactions = append(transactions, blockData.Transactions...)
-		traces = append(traces, blockData.Traces...)
+		if !blockExistenceMap[blockData.Block.Number.String()] {
+			blocks = append(blocks, blockData.Block)
+			logs = append(logs, blockData.Logs...)
+			transactions = append(transactions, blockData.Transactions...)
+			traces = append(traces, blockData.Traces...)
+		}
 	}
 
 	var saveErr error