eth/db_upgrade: refactor for deduplicate data (txLookups)

eth/backend.go

@@ -103,7 +103,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 	if err != nil {
 		return nil, err
 	}
-	stopDbUpgrade := upgradeSequentialKeys(chainDb)
+	stopDbUpgrade := upgradeDeduplicateData(chainDb)
 	chainConfig, genesisHash, genesisErr := core.SetupGenesisBlock(chainDb, config.Genesis)
 	if _, ok := genesisErr.(*params.ConfigCompatError); genesisErr != nil && !ok {
 		return nil, genesisErr

eth/db_upgrade.go

@@ -24,7 +24,6 @@ import (
 
 	"github.com/ubiq/go-ubiq/common"
 	"github.com/ubiq/go-ubiq/core"
-	"github.com/ubiq/go-ubiq/core/types"
 	"github.com/ubiq/go-ubiq/ethdb"
 	"github.com/ubiq/go-ubiq/log"
 	"github.com/ubiq/go-ubiq/rlp"
@@ -46,208 +45,94 @@ func upgradeDeduplicateData(db ethdb.Database) func() error {
 		db.Put(deduplicateData, []byte{42})
 		return nil
 	}
-
+	// Start the deduplication upgrade on a new goroutine
-	log.Warn("Upgrading chain database to use sequential keys")
+	log.Info("Upgrading database to use lookup entries")
-
+	stop := make(chan chan error)
-	stopChn := make(chan struct{})
-	stoppedChn := make(chan struct{})
 
 	go func() {
-		stopFn := func() bool {
+		// Create an iterator to read the entire database and covert old lookup entires
-			select {
+		it := db.(*ethdb.LDBDatabase).NewIterator()
-			case <-time.After(time.Microsecond * 100): // make sure other processes don't get starved
+		defer func() {
-			case <-stopChn:
+			if it != nil {
-				return true
-			}
-			return false
-		}
-
-		err, stopped := upgradeSequentialCanonicalNumbers(db, stopFn)
-		if err == nil && !stopped {
-			err, stopped = upgradeSequentialBlocks(db, stopFn)
-		}
-		if err == nil && !stopped {
-			err, stopped = upgradeSequentialOrphanedReceipts(db, stopFn)
-		}
-		if err == nil && !stopped {
-			log.Info("Database conversion successful")
-			db.Put(useSequentialKeys, []byte{42})
-		}
-		if err != nil {
-			log.Error("Database conversion failed", "err", err)
-		}
-		close(stoppedChn)
-	}()
-
-	return func() {
-		close(stopChn)
-		<-stoppedChn
-	}
-}
-
-// upgradeSequentialCanonicalNumbers reads all old format canonical numbers from
-// the database, writes them in new format and deletes the old ones if successful.
-func upgradeSequentialCanonicalNumbers(db ethdb.Database, stopFn func() bool) (error, bool) {
-	prefix := []byte("block-num-")
-	it := db.(*ethdb.LDBDatabase).NewIterator()
-	defer func() {
-		it.Release()
-	}()
-	it.Seek(prefix)
-	cnt := 0
-	for bytes.HasPrefix(it.Key(), prefix) {
-		keyPtr := it.Key()
-		if len(keyPtr) < 20 {
-			cnt++
-			if cnt%100000 == 0 {
 				it.Release()
-				it = db.(*ethdb.LDBDatabase).NewIterator()
-				it.Seek(keyPtr)
-				log.Info("Converting canonical numbers", "count", cnt)
 			}
-			number := big.NewInt(0).SetBytes(keyPtr[10:]).Uint64()
+		}()
-			newKey := []byte("h12345678n")
-			binary.BigEndian.PutUint64(newKey[1:9], number)
-			if err := db.Put(newKey, it.Value()); err != nil {
-				return err, false
-			}
-			if err := db.Delete(keyPtr); err != nil {
-				return err, false
-			}
-		}
 
-		if stopFn() {
+		var (
-			return nil, true
+			converted uint64
-		}
+			failed    error
-		it.Next()
+		)
-	}
+		for failed == nil && it.Next() {
-	if cnt > 0 {
+			// Skip any entries that don't look like old transaction meta entires (<hash>0x01)
-		log.Info("converted canonical numbers", "count", cnt)
+			key := it.Key()
-	}
+			if len(key) != common.HashLength+1 || key[common.HashLength] != 0x01 {
-	return nil, false
+				continue
-}
+			}
+			// Skip any entries that don't contain metadata (name clash between <hash>0x01 and <some-prefix><hash>)
+			var meta struct {
+				BlockHash  common.Hash
+				BlockIndex uint64
+				Index      uint64
+			}
+			if err := rlp.DecodeBytes(it.Value(), &meta); err != nil {
+				continue
+			}
+			// Skip any already upgraded entries (clash due to <hash> ending with 0x01 (old suffix))
+			hash := key[:common.HashLength]
 
-// upgradeSequentialBlocks reads all old format block headers, bodies, TDs and block
+			if hash[0] == byte('l') {
-// receipts from the database, writes them in new format and deletes the old ones
+				// Potential clash, the "old" `hash` must point to a live transaction.
-// if successful.
+				if tx, _, _, _ := core.GetTransaction(db, common.BytesToHash(hash)); tx == nil || !bytes.Equal(tx.Hash().Bytes(), hash) {
-func upgradeSequentialBlocks(db ethdb.Database, stopFn func() bool) (error, bool) {
+					continue
-	prefix := []byte("block-")
-	it := db.(*ethdb.LDBDatabase).NewIterator()
-	defer func() {
-		it.Release()
-	}()
-	it.Seek(prefix)
-	cnt := 0
-	for bytes.HasPrefix(it.Key(), prefix) {
-		keyPtr := it.Key()
-		if len(keyPtr) >= 38 {
-			cnt++
-			if cnt%10000 == 0 {
-				it.Release()
-				it = db.(*ethdb.LDBDatabase).NewIterator()
-				it.Seek(keyPtr)
-				log.Info("Converting blocks", "count", cnt)
-			}
-			// convert header, body, td and block receipts
-			var keyPrefix [38]byte
-			copy(keyPrefix[:], keyPtr[0:38])
-			hash := keyPrefix[6:38]
-			if err := upgradeSequentialBlockData(db, hash); err != nil {
-				return err, false
-			}
-			// delete old db entries belonging to this hash
-			for bytes.HasPrefix(it.Key(), keyPrefix[:]) {
-				if err := db.Delete(it.Key()); err != nil {
-					return err, false
 				}
-				it.Next()
 			}
-			if err := db.Delete(append([]byte("receipts-block-"), hash...)); err != nil {
+			// Convert the old metadata to a new lookup entry, delete duplicate data
-				return err, false
+			if failed = db.Put(append([]byte("l"), hash...), it.Value()); failed == nil { // Write the new looku entry
+				if failed = db.Delete(hash); failed == nil { // Delete the duplicate transaction data
+					if failed = db.Delete(append([]byte("receipts-"), hash...)); failed == nil { // Delete the duplicate receipt data
+						if failed = db.Delete(key); failed != nil { // Delete the old transaction metadata
+							break
+						}
+					}
+				}
 			}
+			// Bump the conversion counter, and recreate the iterator occasionally to
+			// avoid too high memory consumption.
+			converted++
+			if converted%100000 == 0 {
+				it.Release()
+				it = db.(*ethdb.LDBDatabase).NewIterator()
+				it.Seek(key)
+
+				log.Info("Deduplicating database entries", converted)
+			}
+			// Check for termination, or continue after a bit of a timeout
+			select {
+			case errc := <-stop:
+				errc <- nil
+				return
+			case <-time.After(time.Microsecond * 100):
+			}
+		}
+		// Upgrade finished, mark a such and terminate
+		if failed == nil {
+			log.Info("Database deduplication successful", converted)
+			db.Put(deduplicateData, []byte{42})
 		} else {
-			it.Next()
+			log.Error("Database deduplication failed.", converted, "err", failed)
 		}
+		it.Release()
+		it = nil
 
-		if stopFn() {
+		errc := <-stop
-			return nil, true
+		errc <- failed
-		}
+	}()
+	// Assembly the cancellation callback
+	return func() error {
+		errc := make(chan error)
+		stop <- errc
+		return <-errc
 	}
-	if cnt > 0 {
-		log.Info("Converted blocks", "count", cnt)
-	}
-	return nil, false
-}
-
-// upgradeSequentialOrphanedReceipts removes any old format block receipts from the
-// database that did not have a corresponding block
-func upgradeSequentialOrphanedReceipts(db ethdb.Database, stopFn func() bool) (error, bool) {
-	prefix := []byte("receipts-block-")
-	it := db.(*ethdb.LDBDatabase).NewIterator()
-	defer it.Release()
-	it.Seek(prefix)
-	cnt := 0
-	for bytes.HasPrefix(it.Key(), prefix) {
-		// phase 2 already converted receipts belonging to existing
-		// blocks, just remove if there's anything left
-		cnt++
-		if err := db.Delete(it.Key()); err != nil {
-			return err, false
-		}
-
-		if stopFn() {
-			return nil, true
-		}
-		it.Next()
-	}
-	if cnt > 0 {
-		log.Info("Removed orphaned block receipts", "count", cnt)
-	}
-	return nil, false
-}
-
-// upgradeSequentialBlockData upgrades the header, body, td and block receipts
-// database entries belonging to a single hash (doesn't delete old data).
-func upgradeSequentialBlockData(db ethdb.Database, hash []byte) error {
-	// get old chain data and block number
-	headerRLP, _ := db.Get(append(append([]byte("block-"), hash...), []byte("-header")...))
-	if len(headerRLP) == 0 {
-		return nil
-	}
-	header := new(types.Header)
-	if err := rlp.Decode(bytes.NewReader(headerRLP), header); err != nil {
-		return err
-	}
-	number := header.Number.Uint64()
-	bodyRLP, _ := db.Get(append(append([]byte("block-"), hash...), []byte("-body")...))
-	tdRLP, _ := db.Get(append(append([]byte("block-"), hash...), []byte("-td")...))
-	receiptsRLP, _ := db.Get(append([]byte("receipts-block-"), hash...))
-	// store new hash -> number association
-	encNum := make([]byte, 8)
-	binary.BigEndian.PutUint64(encNum, number)
-	if err := db.Put(append([]byte("H"), hash...), encNum); err != nil {
-		return err
-	}
-	// store new chain data
-	if err := db.Put(append(append([]byte("h"), encNum...), hash...), headerRLP); err != nil {
-		return err
-	}
-	if len(tdRLP) != 0 {
-		if err := db.Put(append(append(append([]byte("h"), encNum...), hash...), []byte("t")...), tdRLP); err != nil {
-			return err
-		}
-	}
-	if len(bodyRLP) != 0 {
-		if err := db.Put(append(append([]byte("b"), encNum...), hash...), bodyRLP); err != nil {
-			return err
-		}
-	}
-	if len(receiptsRLP) != 0 {
-		if err := db.Put(append(append([]byte("r"), encNum...), hash...), receiptsRLP); err != nil {
-			return err
-		}
-	}
-	return nil
 }
 
 func addMipmapBloomBins(db ethdb.Database) (err error) {