cleanup blocks on committer start (#260)

Author: nischitpra

internal/orchestrator/committer.go

@@ -79,6 +79,10 @@ func (c *Committer) Start(ctx context.Context) {
 	interval := time.Duration(c.triggerIntervalMs) * time.Millisecond
 
 	log.Debug().Msgf("Committer running")
+
+	// Clean up staging data before starting the committer
+	c.cleanupStagingData()
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -112,6 +116,28 @@ func (c *Committer) Start(ctx context.Context) {
 	}
 }
 
+func (c *Committer) cleanupStagingData() {
+	// Get the last committed block number from main storage
+	latestCommittedBlockNumber, err := c.storage.MainStorage.GetMaxBlockNumber(c.rpc.GetChainID())
+	if err != nil {
+		log.Error().Msgf("Error getting latest committed block number: %v", err)
+		return
+	}
+
+	if latestCommittedBlockNumber.Sign() == 0 {
+		log.Debug().Msg("No blocks committed yet, skipping staging data cleanup")
+		return
+	}
+
+	// Delete all staging data older than the latest committed block number
+	if err := c.storage.StagingStorage.DeleteOlderThan(c.rpc.GetChainID(), latestCommittedBlockNumber); err != nil {
+		log.Error().Msgf("Error deleting staging data older than %v: %v", latestCommittedBlockNumber, err)
+		return
+	}
+
+	log.Info().Msgf("Deleted staging data older than or equal to %v", latestCommittedBlockNumber)
+}
+
 func (c *Committer) getBlockNumbersToCommit(ctx context.Context) ([]*big.Int, error) {
 	startTime := time.Now()
 	defer func() {

internal/orchestrator/committer_test.go

@@ -404,6 +404,9 @@ func TestStartCommitter(t *testing.T) {
 	mockRPC.EXPECT().GetChainID().Return(chainID)
 	mockMainStorage.EXPECT().GetMaxBlockNumber(chainID).Return(big.NewInt(100), nil)
 
+	// Add expectation for DeleteOlderThan call during cleanup
+	mockStagingStorage.On("DeleteOlderThan", chainID, big.NewInt(100)).Return(nil)
+
 	blockData := []common.BlockData{
 		{Block: common.Block{Number: big.NewInt(101)}},
 		{Block: common.Block{Number: big.NewInt(102)}},
@@ -438,6 +441,9 @@ func TestCommitterRespectsSIGTERM(t *testing.T) {
 	mockRPC.EXPECT().GetChainID().Return(chainID)
 	mockMainStorage.EXPECT().GetMaxBlockNumber(chainID).Return(big.NewInt(100), nil)
 
+	// Add expectation for DeleteOlderThan call during cleanup
+	mockStagingStorage.On("DeleteOlderThan", chainID, big.NewInt(100)).Return(nil)
+
 	blockData := []common.BlockData{
 		{Block: common.Block{Number: big.NewInt(101)}},
 		{Block: common.Block{Number: big.NewInt(102)}},

internal/storage/clickhouse.go

@@ -2114,6 +2114,17 @@ func (c *ClickHouseConnector) GetFullBlockData(chainId *big.Int, blockNumbers []
 	return blockData, nil
 }
 
+func (c *ClickHouseConnector) DeleteOlderThan(chainId *big.Int, blockNumber *big.Int) error {
+	query := fmt.Sprintf(`
+		INSERT INTO %s.block_data (chain_id, block_number, is_deleted)
+		SELECT chain_id, block_number, 1
+		FROM %s.block_data
+		WHERE chain_id = ? AND block_number <= ? AND is_deleted = 0
+		GROUP BY chain_id, block_number
+	`, c.cfg.Database, c.cfg.Database)
+	return c.conn.Exec(context.Background(), query, chainId, blockNumber)
+}
+
 // Helper function to test query generation
 func (c *ClickHouseConnector) TestQueryGeneration(table, columns string, qf QueryFilter) string {
 	return c.buildQuery(table, columns, qf)

internal/storage/connector.go

@@ -83,6 +83,7 @@ type IStagingStorage interface {
 	GetStagingData(qf QueryFilter) (data []common.BlockData, err error)
 	DeleteStagingData(data []common.BlockData) error
 	GetLastStagedBlockNumber(chainId *big.Int, rangeStart *big.Int, rangeEnd *big.Int) (maxBlockNumber *big.Int, err error)
+	DeleteOlderThan(chainId *big.Int, blockNumber *big.Int) error
 }
 
 type IMainStorage interface {

internal/storage/postgres.go

@@ -387,6 +387,12 @@ func (p *PostgresConnector) GetLastStagedBlockNumber(chainId *big.Int, rangeStar
 	return blockNumber, nil
 }
 
+func (p *PostgresConnector) DeleteOlderThan(chainId *big.Int, blockNumber *big.Int) error {
+	query := `DELETE FROM block_data WHERE chain_id = $1 AND block_number <= $2`
+	_, err := p.db.Exec(query, chainId.String(), blockNumber.String())
+	return err
+}
+
 // Close closes the database connection
 func (p *PostgresConnector) Close() error {
 	return p.db.Close()