fix pruner timing issue with batch pruning (#14929)

* fix pruner timing issue with batch pruning

* add changelog entry

* add batchSize and number of slots deleted to debug logs

* fix lint

* prune in small batches in one prune cycle

* remove noisy logs

* fix number of batches to prune and return early if there's nothing to delete

* use context with timeout

* fix lint by ignoring nil err return

beacon-chain/db/kv/blocks_test.go

@@ -359,184 +359,221 @@ func TestStore_DeleteFinalizedBlock(t *testing.T) {
 
 func TestStore_HistoricalDataBeforeSlot(t *testing.T) {
 	slotsPerEpoch := uint64(params.BeaconConfig().SlotsPerEpoch)
-	db := setupDB(t)
 	ctx := context.Background()
 
-	// Save genesis block root
-	require.NoError(t, db.SaveGenesisBlockRoot(ctx, genesisBlockRoot))
+	tests := []struct {
+		name             string
+		batchSize        int
+		numOfEpochs      uint64
+		deleteBeforeSlot uint64
+	}{
+		{
+			name:             "batchSize less than delete range",
+			batchSize:        10,
+			numOfEpochs:      4,
+			deleteBeforeSlot: 25,
+		},
+		{
+			name:             "batchSize greater than delete range",
+			batchSize:        30,
+			numOfEpochs:      4,
+			deleteBeforeSlot: 15,
+		},
+	}
 
-	// Create and save blocks for 4 epochs
-	blks := makeBlocks(t, 0, slotsPerEpoch*4, genesisBlockRoot)
-	require.NoError(t, db.SaveBlocks(ctx, blks))
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			db := setupDB(t)
+			// Save genesis block root
+			require.NoError(t, db.SaveGenesisBlockRoot(ctx, genesisBlockRoot))
 
-	// Mark state validator migration as complete
-	err := db.db.Update(func(tx *bolt.Tx) error {
-		return tx.Bucket(migrationsBucket).Put(migrationStateValidatorsKey, migrationCompleted)
-	})
-	require.NoError(t, err)
+			// Create and save blocks for given epochs
+			blks := makeBlocks(t, 0, slotsPerEpoch*tt.numOfEpochs, genesisBlockRoot)
+			require.NoError(t, db.SaveBlocks(ctx, blks))
 
-	migrated, err := db.isStateValidatorMigrationOver()
-	require.NoError(t, err)
-	require.Equal(t, true, migrated)
+			// Mark state validator migration as complete
+			err := db.db.Update(func(tx *bolt.Tx) error {
+				return tx.Bucket(migrationsBucket).Put(migrationStateValidatorsKey, migrationCompleted)
+			})
+			require.NoError(t, err)
 
-	// Create state summaries and states for each block
-	ss := make([]*ethpb.StateSummary, len(blks))
-	states := make([]state.BeaconState, len(blks))
+			migrated, err := db.isStateValidatorMigrationOver()
+			require.NoError(t, err)
+			require.Equal(t, true, migrated)
 
-	for i, blk := range blks {
-		slot := blk.Block().Slot()
-		r, err := blk.Block().HashTreeRoot()
-		require.NoError(t, err)
+			// Create state summaries and states for each block
+			ss := make([]*ethpb.StateSummary, len(blks))
+			states := make([]state.BeaconState, len(blks))
 
-		// Create and save state summary
-		ss[i] = &ethpb.StateSummary{
-			Slot: slot,
-			Root: r[:],
-		}
+			for i, blk := range blks {
+				slot := blk.Block().Slot()
+				r, err := blk.Block().HashTreeRoot()
+				require.NoError(t, err)
 
-		// Create and save state with validator entries
-		vals := make([]*ethpb.Validator, 2)
-		for j := range vals {
-			vals[j] = &ethpb.Validator{
-				PublicKey:             bytesutil.PadTo([]byte{byte(i*j + 1)}, 48),
-				WithdrawalCredentials: bytesutil.PadTo([]byte{byte(i*j + 2)}, 32),
+				// Create and save state summary
+				ss[i] = &ethpb.StateSummary{
+					Slot: slot,
+					Root: r[:],
+				}
+
+				// Create and save state with validator entries
+				vals := make([]*ethpb.Validator, 2)
+				for j := range vals {
+					vals[j] = &ethpb.Validator{
+						PublicKey:             bytesutil.PadTo([]byte{byte(i*j + 1)}, 48),
+						WithdrawalCredentials: bytesutil.PadTo([]byte{byte(i*j + 2)}, 32),
+					}
+				}
+
+				st, err := util.NewBeaconState(func(state *ethpb.BeaconState) error {
+					state.Validators = vals
+					state.Slot = slot
+					return nil
+				})
+				require.NoError(t, err)
+				require.NoError(t, db.SaveState(ctx, st, r))
+				states[i] = st
+
+				// Verify validator entries are saved to db
+				valsActual, err := db.validatorEntries(ctx, r)
+				require.NoError(t, err)
+				for j, val := range valsActual {
+					require.DeepEqual(t, vals[j], val)
+				}
 			}
-		}
+			require.NoError(t, db.SaveStateSummaries(ctx, ss))
 
-		st, err := util.NewBeaconState(func(state *ethpb.BeaconState) error {
-			state.Validators = vals
-			state.Slot = slot
-			return nil
-		})
-		require.NoError(t, err)
-		require.NoError(t, db.SaveState(ctx, st, r))
-		states[i] = st
-
-		// Verify validator entries are saved to db
-		valsActual, err := db.validatorEntries(ctx, r)
-		require.NoError(t, err)
-		for j, val := range valsActual {
-			require.DeepEqual(t, vals[j], val)
-		}
-	}
-	require.NoError(t, db.SaveStateSummaries(ctx, ss))
-
-	// Verify slot indices exist before deletion
-	err = db.db.View(func(tx *bolt.Tx) error {
-		blockSlotBkt := tx.Bucket(blockSlotIndicesBucket)
-		stateSlotBkt := tx.Bucket(stateSlotIndicesBucket)
-
-		for i := uint64(0); i < slotsPerEpoch; i++ {
-			slot := bytesutil.SlotToBytesBigEndian(primitives.Slot(i + 1))
-			assert.NotNil(t, blockSlotBkt.Get(slot), "Expected block slot index to exist")
-			assert.NotNil(t, stateSlotBkt.Get(slot), "Expected state slot index to exist", i)
-		}
-		return nil
-	})
-	require.NoError(t, err)
-
-	// Delete data before slot at epoch 1
-	require.NoError(t, db.DeleteHistoricalDataBeforeSlot(ctx, primitives.Slot(slotsPerEpoch)))
-
-	// Verify blocks from epoch 0 are deleted
-	for i := uint64(0); i < slotsPerEpoch; i++ {
-		root, err := blks[i].Block().HashTreeRoot()
-		require.NoError(t, err)
-
-		// Check block is deleted
-		retrievedBlocks, err := db.BlocksBySlot(ctx, primitives.Slot(i))
-		require.NoError(t, err)
-		assert.Equal(t, 0, len(retrievedBlocks))
-
-		// Verify block does not exist
-		assert.Equal(t, false, db.HasBlock(ctx, root))
-
-		// Verify block parent root does not exist
-		err = db.db.View(func(tx *bolt.Tx) error {
-			require.Equal(t, 0, len(tx.Bucket(blockParentRootIndicesBucket).Get(root[:])))
-			return nil
-		})
-		require.NoError(t, err)
-
-		// Verify state is deleted
-		hasState := db.HasState(ctx, root)
-		assert.Equal(t, false, hasState)
-
-		// Verify state summary is deleted
-		hasSummary := db.HasStateSummary(ctx, root)
-		assert.Equal(t, false, hasSummary)
-
-		// Verify validator hashes for block roots are deleted
-		err = db.db.View(func(tx *bolt.Tx) error {
-			assert.Equal(t, 0, len(tx.Bucket(blockRootValidatorHashesBucket).Get(root[:])))
-			return nil
-		})
-		require.NoError(t, err)
-	}
-
-	// Verify slot indices are deleted
-	err = db.db.View(func(tx *bolt.Tx) error {
-		blockSlotBkt := tx.Bucket(blockSlotIndicesBucket)
-		stateSlotBkt := tx.Bucket(stateSlotIndicesBucket)
-
-		for i := uint64(0); i < slotsPerEpoch; i++ {
-			slot := bytesutil.SlotToBytesBigEndian(primitives.Slot(i + 1))
-			assert.Equal(t, 0, len(blockSlotBkt.Get(slot)), fmt.Sprintf("Expected block slot index to be deleted, slot: %d", slot))
-			assert.Equal(t, 0, len(stateSlotBkt.Get(slot)), fmt.Sprintf("Expected state slot index to be deleted, slot: %d", slot))
-		}
-		return nil
-	})
-	require.NoError(t, err)
-
-	// Verify blocks from epochs 1-3 still exist
-	for i := slotsPerEpoch; i < slotsPerEpoch*4; i++ {
-		root, err := blks[i].Block().HashTreeRoot()
-		require.NoError(t, err)
-
-		// Verify block exists
-		assert.Equal(t, true, db.HasBlock(ctx, root))
-
-		// Verify remaining block parent root exists, except last slot since we store parent roots of each block.
-		if i < slotsPerEpoch*4-1 {
+			// Verify slot indices exist before deletion
 			err = db.db.View(func(tx *bolt.Tx) error {
-				require.NotNil(t, tx.Bucket(blockParentRootIndicesBucket).Get(root[:]), fmt.Sprintf("Expected block parent index to be deleted, slot: %d", i))
+				blockSlotBkt := tx.Bucket(blockSlotIndicesBucket)
+				stateSlotBkt := tx.Bucket(stateSlotIndicesBucket)
+
+				for i := uint64(0); i < uint64(tt.deleteBeforeSlot); i++ {
+					slot := bytesutil.SlotToBytesBigEndian(primitives.Slot(i + 1))
+					assert.NotNil(t, blockSlotBkt.Get(slot), "Expected block slot index to exist")
+					assert.NotNil(t, stateSlotBkt.Get(slot), "Expected state slot index to exist", i)
+				}
 				return nil
 			})
 			require.NoError(t, err)
-		}
 
-		// Verify state exists
-		hasState := db.HasState(ctx, root)
-		assert.Equal(t, true, hasState)
+			// Delete data before slot
+			slotsDeleted, err := db.DeleteHistoricalDataBeforeSlot(ctx, primitives.Slot(tt.deleteBeforeSlot), tt.batchSize)
+			require.NoError(t, err)
 
-		// Verify state summary exists
-		hasSummary := db.HasStateSummary(ctx, root)
-		assert.Equal(t, true, hasSummary)
+			var startSlotDeleted, endSlotDeleted uint64
+			if tt.batchSize >= int(tt.deleteBeforeSlot) {
+				startSlotDeleted = 1
+				endSlotDeleted = tt.deleteBeforeSlot
+			} else {
+				startSlotDeleted = tt.deleteBeforeSlot - uint64(tt.batchSize) + 1
+				endSlotDeleted = tt.deleteBeforeSlot
+			}
 
-		// Verify slot indices still exist
-		err = db.db.View(func(tx *bolt.Tx) error {
-			blockSlotBkt := tx.Bucket(blockSlotIndicesBucket)
-			stateSlotBkt := tx.Bucket(stateSlotIndicesBucket)
+			require.Equal(t, endSlotDeleted-startSlotDeleted+1, uint64(slotsDeleted))
 
-			slot := bytesutil.SlotToBytesBigEndian(primitives.Slot(i + 1))
-			assert.NotNil(t, blockSlotBkt.Get(slot), "Expected block slot index to exist")
-			assert.NotNil(t, stateSlotBkt.Get(slot), "Expected state slot index to exist")
-			return nil
+			// Verify blocks before given slot/batch are deleted
+			for i := startSlotDeleted; i < endSlotDeleted; i++ {
+				root, err := blks[i].Block().HashTreeRoot()
+				require.NoError(t, err)
+
+				// Check block is deleted
+				retrievedBlocks, err := db.BlocksBySlot(ctx, primitives.Slot(i))
+				require.NoError(t, err)
+				assert.Equal(t, 0, len(retrievedBlocks), fmt.Sprintf("Expected %d blocks, got %d for slot %d", 0, len(retrievedBlocks), i))
+
+				// Verify block does not exist
+				assert.Equal(t, false, db.HasBlock(ctx, root), fmt.Sprintf("Expected block index to not exist for slot %d", i))
+
+				// Verify block parent root does not exist
+				err = db.db.View(func(tx *bolt.Tx) error {
+					require.Equal(t, 0, len(tx.Bucket(blockParentRootIndicesBucket).Get(root[:])))
+					return nil
+				})
+				require.NoError(t, err)
+
+				// Verify state is deleted
+				hasState := db.HasState(ctx, root)
+				assert.Equal(t, false, hasState)
+
+				// Verify state summary is deleted
+				hasSummary := db.HasStateSummary(ctx, root)
+				assert.Equal(t, false, hasSummary)
+
+				// Verify validator hashes for block roots are deleted
+				err = db.db.View(func(tx *bolt.Tx) error {
+					assert.Equal(t, 0, len(tx.Bucket(blockRootValidatorHashesBucket).Get(root[:])))
+					return nil
+				})
+				require.NoError(t, err)
+			}
+
+			// Verify slot indices are deleted
+			err = db.db.View(func(tx *bolt.Tx) error {
+				blockSlotBkt := tx.Bucket(blockSlotIndicesBucket)
+				stateSlotBkt := tx.Bucket(stateSlotIndicesBucket)
+
+				for i := startSlotDeleted; i < endSlotDeleted; i++ {
+					slot := bytesutil.SlotToBytesBigEndian(primitives.Slot(i + 1))
+					assert.Equal(t, 0, len(blockSlotBkt.Get(slot)), fmt.Sprintf("Expected block slot index to be deleted, slot: %d", slot))
+					assert.Equal(t, 0, len(stateSlotBkt.Get(slot)), fmt.Sprintf("Expected state slot index to be deleted, slot: %d", slot))
+				}
+				return nil
+			})
+			require.NoError(t, err)
+
+			// Verify blocks from expectedLastDeletedSlot till numEpochs still exist
+			for i := endSlotDeleted; i < slotsPerEpoch*tt.numOfEpochs; i++ {
+				root, err := blks[i].Block().HashTreeRoot()
+				require.NoError(t, err)
+
+				// Verify block exists
+				assert.Equal(t, true, db.HasBlock(ctx, root))
+
+				// Verify remaining block parent root exists, except last slot since we store parent roots of each block.
+				if i < slotsPerEpoch*tt.numOfEpochs-1 {
+					err = db.db.View(func(tx *bolt.Tx) error {
+						require.NotNil(t, tx.Bucket(blockParentRootIndicesBucket).Get(root[:]), fmt.Sprintf("Expected block parent index to be deleted, slot: %d", i))
+						return nil
+					})
+					require.NoError(t, err)
+				}
+
+				// Verify state exists
+				hasState := db.HasState(ctx, root)
+				assert.Equal(t, true, hasState)
+
+				// Verify state summary exists
+				hasSummary := db.HasStateSummary(ctx, root)
+				assert.Equal(t, true, hasSummary)
+
+				// Verify slot indices still exist
+				err = db.db.View(func(tx *bolt.Tx) error {
+					blockSlotBkt := tx.Bucket(blockSlotIndicesBucket)
+					stateSlotBkt := tx.Bucket(stateSlotIndicesBucket)
+
+					slot := bytesutil.SlotToBytesBigEndian(primitives.Slot(i + 1))
+					assert.NotNil(t, blockSlotBkt.Get(slot), "Expected block slot index to exist")
+					assert.NotNil(t, stateSlotBkt.Get(slot), "Expected state slot index to exist")
+					return nil
+				})
+				require.NoError(t, err)
+
+				// Verify validator entries still exist
+				valsActual, err := db.validatorEntries(ctx, root)
+				require.NoError(t, err)
+				assert.NotNil(t, valsActual)
+
+				// Verify remaining validator hashes for block roots exists
+				err = db.db.View(func(tx *bolt.Tx) error {
+					assert.NotNil(t, tx.Bucket(blockRootValidatorHashesBucket).Get(root[:]))
+					return nil
+				})
+				require.NoError(t, err)
+			}
 		})
-		require.NoError(t, err)
-
-		// Verify validator entries still exist
-		valsActual, err := db.validatorEntries(ctx, root)
-		require.NoError(t, err)
-		assert.NotNil(t, valsActual)
-
-		// Verify remaining validator hashes for block roots exists
-		err = db.db.View(func(tx *bolt.Tx) error {
-			assert.NotNil(t, tx.Bucket(blockRootValidatorHashesBucket).Get(root[:]))
-			return nil
-		})
-		require.NoError(t, err)
 	}
+
 }
 
 func TestStore_GenesisBlock(t *testing.T) {