cmd/geth: Refactor state migration initialization

Using the freezer to init the database was fine when we were using
a local ancients database, but is very slow with the s3 freezer.
This refactors to pull those records across from the old database
rather than reconstructing them from blocks.

Author: AusIV

cmd/geth/chaincmd.go

@@ -895,6 +895,9 @@ func syncState(root common.Hash, srcDb state.Database, newDb ethdb.Database) <-c
 					defer wg.Done()
 					for r := range ch {
 						r.data, r.err = srcDb.TrieDB().Node(r.hash)
+						if r.err != nil {
+							log.Warn("Error processing hash", "hash", r.hash, "err", r.err)
+						}
 						popCh <- r
 					}
 				}(&wg)
@@ -911,17 +914,20 @@ func syncState(root common.Hash, srcDb state.Database, newDb ethdb.Database) <-c
 			}()
 			for r := range popCh {
 				if r.err != nil {
+					log.Crit("trie node error", "err", r.err)
 					errCh <- r.err
 					return
 				}
 				results[r.i] = trie.SyncResult{Hash: r.hash, Data: r.data}
 			}
 			if _, index, err := sched.Process(results); err != nil {
+				log.Crit("trie processing error", "err", err)
 				errCh <- fmt.Errorf("failed to process result #%d: %v", index, err)
 				return
 			}
 			batch := newDb.NewBatch()
 			if err := sched.Commit(batch); err != nil {
+				log.Crit("commit error", "err", err)
 				errCh <- fmt.Errorf("failed to commit data: %v", err)
 				return
 			}
@@ -976,26 +982,92 @@ func migrateState(ctx *cli.Context) error {
     return fmt.Errorf("Usage: migrateState [ancients] [oldLeveldb] [newLeveldb] [?kafkaTopic]")
   }
 	newDb, err := rawdb.NewLevelDBDatabaseWithFreezer(ctx.Args()[2], 16, 16, ctx.Args()[0], "new")
-	if err != nil { return err }
+	if err != nil {
+		log.Crit("Error new opening database")
+		return err
+	}
 	frozen, err := newDb.Ancients()
-	if err != nil { return err }
+	if err != nil {
+		log.Crit("Error getting ancients")
+		return err
+	}
 	if frozen == 0 {
 		return fmt.Errorf("Freezer is empty")
 	}
 	oldDb, err := rawdb.NewLevelDBDatabase(ctx.Args()[1], 16, 16, "old")
-	if err != nil { return err }
+	if err != nil {
+		log.Crit("Error old opening database")
+		return err
+	}
 	if len(ctx.Args()) == 4 {
 		key := fmt.Sprintf("cdc-log-%v-offset", ctx.Args()[3])
 		offset, err := oldDb.Get([]byte(key))
-		if err != nil { return err }
+		if err != nil {
+			log.Crit("Error getting offset from database")
+			return err
+		}
 		if err := newDb.Put([]byte(key), offset); err != nil { return err }
 		log.Info("Copied offset", "key", key)
 	}
 	start := time.Now()
 	ancientErrCh := make(chan error, 1)
 	if os.Getenv("SKIP_INIT_FREEZER") != "true" {
 		go func() {
-			rawdb.InitDatabaseFromFreezer(newDb)
+			it := oldDb.NewIterator([]byte("l"), nil)
+			defer it.Release()
+			batch := newDb.NewBatch()
+			for it.Next() {
+				if err := batch.Put(it.Key(), it.Value()); err != nil {
+					ancientErrCh <- err
+					return
+				}
+				if batch.ValueSize() > ethdb.IdealBatchSize {
+					if err := batch.Write(); err != nil {
+						ancientErrCh <- err
+						return
+					}
+					batch.Reset()
+				}
+			}
+			if err := it.Error(); err != nil {
+				ancientErrCh <- err
+				return
+			}
+			headerIt := oldDb.NewIterator([]byte("H"), nil)
+			defer headerIt.Release()
+			for headerIt.Next() {
+				if err := batch.Put(headerIt.Key(), headerIt.Value()); err != nil {
+					ancientErrCh <- err
+					return
+				}
+				if batch.ValueSize() > ethdb.IdealBatchSize {
+					if err := batch.Write(); err != nil {
+						ancientErrCh <- err
+						return
+					}
+					batch.Reset()
+				}
+			}
+			if err := batch.Write(); err != nil {
+				ancientErrCh <- err
+				return
+			}
+			if err := headerIt.Error(); err != nil {
+				ancientErrCh <- err
+				return
+			}
+			blockNo, err := newDb.Ancients()
+			if err != nil {
+				ancientErrCh <- err
+				return
+			}
+			hash := rawdb.ReadCanonicalHash(newDb, blockNo)
+
+			rawdb.WriteHeadHeaderHash(newDb, hash)
+			rawdb.WriteHeadFastBlockHash(newDb, hash)
+
+
+			// rawdb.InitDatabaseFromFreezer(newDb)
 			ancientErrCh <- nil
 			log.Info("Initialized from freezer", "elapsed", time.Since(start))
 		}()
@@ -1007,12 +1079,16 @@ func migrateState(ctx *cli.Context) error {
 	genesisHash := rawdb.ReadCanonicalHash(oldDb, 0)
 	block := rawdb.ReadBlock(oldDb, genesisHash, 0)
 
-	latestBlockHash := rawdb.ReadHeadFastBlockHash(oldDb) // Find the latest blockhash migrated to the new database
+	latestBlockHash := rawdb.ReadHeadBlockHash(oldDb) // Find the latest blockhash migrated to the new database
 	if latestBlockHash == (common.Hash{}) {
 		return fmt.Errorf("Source block hash empty")
 	}
 	latestHeaderNumber := rawdb.ReadHeaderNumber(oldDb, latestBlockHash)
-	latestBlock := rawdb.ReadBlock(newDb, latestBlockHash, *latestHeaderNumber)
+	latestBlock := rawdb.ReadBlock(oldDb, latestBlockHash, *latestHeaderNumber)
+	for _, err := srcDb.TrieDB().Node(latestBlock.Root()); err != nil ; {
+		latestBlock = rawdb.ReadBlock(oldDb, latestBlock.ParentHash(), latestBlock.NumberU64() - 1)
+	}
+
 
 	log.Info("Syncing genesis block state", "hash", block.Hash(), "root", block.Root())
 	genesisErrCh := syncState(block.Root(), srcDb, newDb)
@@ -1063,8 +1139,14 @@ func migrateState(ctx *cli.Context) error {
 		}
 	}
 
-	if err := <-genesisErrCh; err != nil { return err }
-	if err := <-latestErrCh; err != nil { return err }
+	if err := <-genesisErrCh; err != nil {
+		log.Crit("error syncing genesis")
+		return err
+	}
+	if err := <-latestErrCh; err != nil {
+		log.Crit("error syncing latest")
+		return err
+	}
 	rawdb.WriteHeadBlockHash(newDb, block.Hash())
 	rawdb.WriteHeadHeaderHash(newDb, block.Hash())
 	rawdb.WriteHeadFastBlockHash(newDb, block.Hash())