State fixes and improvements (#78)

* Sync [wip]

* Accounts storage: fix filterHistory()

These changes must be written directly to DB, because deleting
outdated records always makes sense.

* BlockReadWriter: do not remove tx IDs in rollback

See comment.

* Add option to apply from certain height

Successfully tested possibility to continue importing after Ctrl-C termination.

* Accounts storage: do not store more than ROLLBACK_MAX blocks

* GO-61: Reduntan mutex removed (#75)

* GO-63: Correct termination, fixed binary search, fixed error messages and return codes (#76)

* Rollback(): remove transaction IDs correctly

Remove separate tx IDs file. Use main blockchain file to get IDs to remove.

* Use camelCase for constants

* Importer: use fixed-size byte array for block

* Combine all related packages into single state

.gitignore

@@ -15,6 +15,9 @@
 *.swo
 *.swp
 
+# Profiling
+*.prof
+
 .idea/
 vendor/
 build/

cmd/importer/importer.go

@@ -15,17 +15,22 @@ import (
 var (
 	blockchainPath = flag.String("blockchain-path", "", "Path to binary blockchain file.")
 	balancesPath   = flag.String("balances-path", "", "Path to JSON with correct balances after applying blocks.")
+	dataDirPath    = flag.String("data-path", "", "Path to directory with previously created state.")
 	nBlocks        = flag.Int("blocks-number", 1000, "Number of blocks to import.")
 )
 
 func main() {
 	flag.Parse()
-	if len(*blockchainPath) == 0 {
+	if *blockchainPath == "" {
 		log.Fatalf("You must specify blockchain-path option.")
 	}
-	dataDir, err := ioutil.TempDir(os.TempDir(), "dataDir")
-	if err != nil {
-		log.Fatalf("Faied to create temp dir for data: %v\n", err)
+	dataDir := *dataDirPath
+	if dataDir == "" {
+		tempDir, err := ioutil.TempDir(os.TempDir(), "dataDir")
+		if err != nil {
+			log.Fatalf("Faied to create temp dir for data: %v\n", err)
+		}
+		dataDir = tempDir
 	}
 	manager, err := state.NewStateManager(dataDir, state.DefaultBlockStorageParams())
 	if err != nil {
@@ -36,13 +41,19 @@ func main() {
 		if err := manager.Close(); err != nil {
 			log.Fatalf("Failed to close StateManager: %v\n", err)
 		}
-		if err := os.RemoveAll(dataDir); err != nil {
-			log.Fatalf("Failed to clean data dir: %v\n", err)
+		if *dataDirPath == "" {
+			if err := os.RemoveAll(dataDir); err != nil {
+				log.Fatalf("Failed to clean data dir: %v\n", err)
+			}
 		}
 	}()
 
+	height, err := manager.Height()
+	if err != nil {
+		log.Fatalf("Failed to get current height: %v\n", err)
+	}
 	start := time.Now()
-	if err := importer.ApplyFromFile(manager, *blockchainPath, *nBlocks, false); err != nil {
+	if err := importer.ApplyFromFile(manager, *blockchainPath, uint64(*nBlocks), height, false); err != nil {
 		log.Fatalf("Failed to apply blocks: %v\n", err)
 	}
 	elapsed := time.Since(start)

pkg/importer/importer.go

@@ -12,22 +12,26 @@ import (
 	"github.com/wavesplatform/gowaves/pkg/proto"
 )
 
+const (
+	maxBlockSize = 2 * 1024 * 1024
+)
+
 type State interface {
 	AcceptAndVerifyBlockBinary(block []byte, initialisation bool) error
 	GetBlockByHeight(height uint64) (*proto.Block, error)
-	WavesAddressesNumber() (uint64, error)
+	AddressesNumber() (uint64, error)
 	AccountBalance(addr proto.Address, asset []byte) (uint64, error)
 }
 
-func ApplyFromFile(st State, blockchainPath string, nBlocks int, checkBlocks bool) error {
+func ApplyFromFile(st State, blockchainPath string, nBlocks, startHeight uint64, checkBlocks bool) error {
 	blockchain, err := os.Open(blockchainPath)
 	if err != nil {
 		return errors.Errorf("failed to open blockchain file: %v\n", err)
 	}
 	sb := make([]byte, 4)
-	buf := make([]byte, 2*1024*1024)
+	var buf [maxBlockSize]byte
 	r := bufio.NewReader(blockchain)
-	for i := 0; i < nBlocks; i++ {
+	for height := uint64(0); height < nBlocks; height++ {
 		if _, err := io.ReadFull(r, sb); err != nil {
 			return err
 		}
@@ -36,20 +40,22 @@ func ApplyFromFile(st State, blockchainPath string, nBlocks int, checkBlocks boo
 		if _, err := io.ReadFull(r, block); err != nil {
 			return err
 		}
-		if err := st.AcceptAndVerifyBlockBinary(block, true); err != nil {
-			return err
-		}
-		if checkBlocks {
-			savedBlock, err := st.GetBlockByHeight(uint64(i))
-			if err != nil {
-				return err
-			}
-			savedBlockBytes, err := savedBlock.MarshalBinary()
-			if err != nil {
+		if height >= startHeight {
+			if err := st.AcceptAndVerifyBlockBinary(block, true); err != nil {
 				return err
 			}
-			if bytes.Compare(block, savedBlockBytes) != 0 {
-				return errors.New("accepted and returned blocks differ\n")
+			if checkBlocks {
+				savedBlock, err := st.GetBlockByHeight(height)
+				if err != nil {
+					return err
+				}
+				savedBlockBytes, err := savedBlock.MarshalBinary()
+				if err != nil {
+					return err
+				}
+				if bytes.Compare(block, savedBlockBytes) != 0 {
+					return errors.New("accepted and returned blocks differ\n")
+				}
 			}
 		}
 	}
@@ -69,7 +75,7 @@ func CheckBalances(st State, balancesPath string) error {
 	if err := jsonParser.Decode(&state); err != nil {
 		return errors.Errorf("failed to decode state: %v\n", err)
 	}
-	addressesNumber, err := st.WavesAddressesNumber()
+	addressesNumber, err := st.AddressesNumber()
 	if err != nil {
 		return errors.Errorf("failed to get number of waves addresses: %v\n", err)
 	}

pkg/keyvalue/common.go

@@ -3,9 +3,11 @@ package keyvalue
 type KeyValue interface {
 	Has(key []byte) (bool, error)
 	Put(key, val []byte) error
+	PutDirectly(key, val []byte) error
 	Get(key []byte) ([]byte, error)
 	Delete(key []byte) error
 	Flush() error
+	Close() error
 }
 
 type Iterator interface {

pkg/keyvalue/leveldb.go

@@ -35,6 +35,13 @@ func (k *KeyVal) Delete(key []byte) error {
 	return k.db.Delete(key, nil)
 }
 
+func (k *KeyVal) PutDirectly(key, val []byte) error {
+	if err := k.db.Put(key, val, nil); err != nil {
+		return err
+	}
+	return nil
+}
+
 func (k *KeyVal) Put(key, val []byte) error {
 	if k.batch != nil {
 		k.batch.Put(key, val)
@@ -64,3 +71,7 @@ func (k *KeyVal) NewKeyIterator(prefix []byte) (Iterator, error) {
 		return k.db.NewIterator(nil, nil), nil
 	}
 }
+
+func (k *KeyVal) Close() error {
+	return k.db.Close()
+}