cleanup blocks on committer start

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/storage/clickhouse.go

@@ -2114,6 +2114,53 @@ func (c *ClickHouseConnector) GetFullBlockData(chainId *big.Int, blockNumbers []
 	return blockData, nil
 }
 
+func (c *ClickHouseConnector) DeleteOlderThan(chainId *big.Int, blockNumber *big.Int) error {
+	// First, get all the block numbers that need to be deleted
+	query := fmt.Sprintf(`
+		SELECT DISTINCT chain_id, block_number
+		FROM %s.block_data 
+		WHERE chain_id = ? AND block_number <= ? AND is_deleted = 0
+	`, c.cfg.Database)
+
+	rows, err := c.conn.Query(context.Background(), query, chainId, blockNumber)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	// Prepare batch for deletion
+	deleteQuery := fmt.Sprintf(`
+		INSERT INTO %s.block_data (
+			chain_id, block_number, is_deleted
+		) VALUES (?, ?, ?)
+	`, c.cfg.Database)
+
+	batch, err := c.conn.PrepareBatch(context.Background(), deleteQuery)
+	if err != nil {
+		return err
+	}
+	defer batch.Close()
+
+	// Add each block to the deletion batch
+	for rows.Next() {
+		var chainIdVal, blockNumberVal *big.Int
+		if err := rows.Scan(&chainIdVal, &blockNumberVal); err != nil {
+			return err
+		}
+
+		err := batch.Append(
+			chainIdVal,
+			blockNumberVal,
+			1, // is_deleted = 1
+		)
+		if err != nil {
+			return err
+		}
+	}
+
+	return batch.Send()
+}
+
 // 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()