test fix

block/pruning.go

@@ -16,28 +16,27 @@ func (m *Manager) PruneBlocks(retainHeight uint64) {
 	pruned, err := m.P2PClient.RemoveBlocks(context.Background(), m.State.BaseBlocksyncHeight, retainHeight)
 	if err != nil {
 		m.logger.Error("pruning blocksync store", "retain_height", retainHeight, "err", err)
+	} else {
+		m.logger.Debug("blocksync store pruned", "from", m.State.BaseBlocksyncHeight, "to", retainHeight, "pruned", pruned)
+		m.State.BaseBlocksyncHeight = retainHeight
 	}
-	m.logger.Debug("blocksync store pruned", "from", m.State.BaseBlocksyncHeight, "to", retainHeight, "pruned", pruned)
-	m.State.BaseBlocksyncHeight = retainHeight
-
 	// prune blocks from indexer store
 	pruned, err = m.IndexerService.Prune(m.State.BaseIndexerHeight, retainHeight)
 	if err != nil {
 		m.logger.Error("pruning indexer", "retain_height", retainHeight, "err", err)
+	} else {
+		m.logger.Debug("indexer store pruned", "from", m.State.BaseIndexerHeight, "to", retainHeight, "pruned", pruned)
+		m.State.BaseIndexerHeight = retainHeight
 	}
-
-	m.logger.Debug("indexer store pruned", "from", m.State.BaseIndexerHeight, "to", retainHeight, "pruned", pruned)
-	m.State.BaseIndexerHeight = retainHeight
-
 	// prune blocks from dymint store
 	pruned, err = m.Store.PruneStore(m.State.BaseHeight, retainHeight, m.logger)
 	if err != nil {
 		m.logger.Error("pruning block store", "retain_height", retainHeight, "err", err)
+	} else {
+		m.logger.Debug("dymint store pruned", "from", m.State.BaseHeight, "to", retainHeight, "pruned", pruned)
+		m.State.BaseHeight = retainHeight
 	}
 
-	m.logger.Debug("dymint store pruned", "from", m.State.BaseHeight, "to", retainHeight, "pruned", pruned)
-	m.State.BaseHeight = retainHeight
-
 	// store state with base heights updates
 	_, err = m.Store.SaveState(m.State, nil)
 	if err != nil {

indexers/blockindexer/kv/kv.go

@@ -535,8 +535,7 @@ func (idx *BlockerIndexer) Prune(from, to uint64, logger log.Logger) (uint64, er
 	if to <= from {
 		return 0, fmt.Errorf("to height must be greater than from height: to: %d: from: %d: %w", to, from, gerrc.ErrInvalidArgument)
 	}
-	blocksPruned, err := idx.pruneBlocks(from, to, logger)
-	return blocksPruned, err
+	return idx.pruneBlocks(from, to, logger)
 }
 
 func (idx *BlockerIndexer) pruneBlocks(from, to uint64, logger log.Logger) (uint64, error) {
@@ -553,14 +552,6 @@ func (idx *BlockerIndexer) pruneBlocks(from, to uint64, logger log.Logger) (uint
 	}
 
 	for h := int64(from); h < int64(to); h++ {
-		ok, err := idx.Has(h)
-		if err != nil {
-			logger.Debug("pruning block indexer checking height", "err", err)
-			continue
-		}
-		if !ok {
-			continue
-		}
 		key, err := heightKey(h)
 		if err != nil {
 			logger.Debug("pruning block indexer getting height key", "err", err)
@@ -570,11 +561,14 @@ func (idx *BlockerIndexer) pruneBlocks(from, to uint64, logger log.Logger) (uint
 			logger.Debug("pruning block indexer deleting height key", "err", err)
 			continue
 		}
+
+		// we update here to avoid acccumulating batches in case pruneEvents fails
+		pruned++
+
 		if err := idx.pruneEvents(h, batch); err != nil {
 			logger.Debug("pruning block indexer events", "err", err)
 			continue
 		}
-		pruned++
 
 		// flush every 1000 blocks to avoid batches becoming too large
 		if pruned%1000 == 0 && pruned > 0 {

indexers/txindex/indexer_service_test.go

@@ -86,7 +86,7 @@ func TestIndexerServiceIndexesBlocks(t *testing.T) {
 
 	txPruned, err := txIndexer.Prune(1, 2, log.NewNopLogger())
 	require.NoError(t, err)
-	expectedTxPruned := uint64(2)
+	expectedTxPruned := uint64(1)
 	require.Equal(t, expectedTxPruned, txPruned)
 
 }

indexers/txindex/kv/kv.go

@@ -595,27 +595,31 @@ func (txi *TxIndex) pruneTxs(from, to uint64) (uint64, error) {
 		it := txi.store.PrefixIterator(prefixForHeight(int64(h)))
 		defer it.Discard()
 
+		if err := txi.pruneEvents(h, batch); err != nil {
+			continue
+		}
+		pruned++
+
 		for ; it.Valid(); it.Next() {
 			if err := batch.Delete(it.Key()); err != nil {
 				continue
 			}
+
 			if err := batch.Delete(it.Value()); err != nil {
 				continue
 			}
-			if err := txi.pruneEvents(h, batch); err != nil {
-				continue
-			}
-			pruned++
-			// flush every 1000 txs to avoid batches becoming too large
-			if pruned%1000 == 0 && pruned > 0 {
-				err := flush(batch, int64(h))
-				if err != nil {
-					return 0, err
-				}
-				batch.Discard()
-				batch = txi.store.NewBatch()
+		}
+
+		// flush every 1000 txs to avoid batches becoming too large
+		if pruned%1000 == 0 && pruned > 0 {
+			err := flush(batch, int64(h))
+			if err != nil {
+				return 0, err
 			}
+			batch.Discard()
+			batch = txi.store.NewBatch()
 		}
+
 	}
 
 	err := flush(batch, int64(to))

p2p/block_sync.go

@@ -113,5 +113,5 @@ func (blocksync *BlockSync) LoadBlock(ctx context.Context, cid cid.Cid) (BlockDa
 
 // RemoveBlock removes the block from the DAGservice.
 func (blocksync *BlockSync) DeleteBlock(ctx context.Context, cid cid.Cid) error {
-	return blocksync.dsrv.Remove(ctx, cid)
+	return blocksync.dsrv.DeleteBlock(ctx, cid)
 }

p2p/block_sync_dag.go

@@ -112,6 +112,24 @@ func (bsDagService *BlockSyncDagService) LoadBlock(ctx context.Context, cid cid.
 	return data, nil
 }
 
+func (bsDagService *BlockSyncDagService) DeleteBlock(ctx context.Context, cid cid.Cid) error {
+
+	// first it gets the root node
+	root, err := bsDagService.Get(ctx, cid)
+	if err != nil {
+		return err
+	}
+
+	// then it iterates all the cids to remove them from the block store
+	for _, l := range root.Links() {
+		err := bsDagService.Remove(ctx, l.Cid)
+		if err != nil {
+			return nil
+		}
+	}
+	return nil
+}
+
 // dagReader is used to read the DAG (all the block chunks) from the root (IPLD) node
 func dagReader(root ipld.Node, ds ipld.DAGService) (io.Reader, error) {
 	ctx := context.Background()