8-to-9: Pre-fill backup file. Reduce size. Better syncing.

Author: hsanjuan

ipfs-8-to-9/migration/migration.go

@@ -9,12 +9,12 @@ import (
 	"io"
 	"os"
 	"path/filepath"
-	"strings"
 
 	migrate "github.com/ipfs/fs-repo-migrations/go-migrate"
 	lock "github.com/ipfs/fs-repo-migrations/ipfs-1-to-2/repolock"
 	"github.com/ipfs/fs-repo-migrations/mfsr"
 	"github.com/ipfs/go-filestore"
+	dshelp "github.com/ipfs/go-ipfs-ds-help"
 
 	log "github.com/ipfs/fs-repo-migrations/stump"
 	ds "github.com/ipfs/go-datastore"
@@ -124,34 +124,47 @@ func (m Migration) Apply(opts migrate.Options) error {
 	}
 	defer f.Close()
 	buf := bufio.NewWriter(f)
-	defer buf.Flush()
 
-	// Will be closed by cidSwapper when it finish writing.
 	swapCh := make(chan Swap, 1000)
 
 	writingDone := make(chan struct{})
 	go func() {
 		for sw := range swapCh {
-			fmt.Fprint(buf, sw.Old.String()+","+sw.New.String()+"\n")
+			// Only write the Old string (a CID). We can derive
+			// the multihash from it.
+			fmt.Fprint(buf, sw.Old.String(), "\n")
 		}
 		close(writingDone)
 	}()
 
+	// Add all the keys to migrate to the backup file
 	for _, prefix := range migrationPrefixes {
-		log.VLog("  - migrating keys for prefix %s", prefix)
+		log.VLog("  - Adding keys in prefix %s to backup file", prefix)
 		cidSwapper := CidSwapper{Prefix: prefix, Store: dstore, SwapCh: swapCh}
-		total, err := cidSwapper.Run()
+		total, err := cidSwapper.Run(true) // DRY RUN
 		if err != nil {
 			close(swapCh)
 			log.Error(err)
 			return err
 		}
-		log.Log("%d CIDv1 keys swapped to raw multihashes for %s", total, prefix)
+		log.Log("%d CIDv1 keys added to backup file for %s", total, prefix)
 	}
 	close(swapCh)
-	// Wait for our writing to finish before doing the final flush
-	// (deferred).
+	// Wait for our writing to finish before doing the flushing.
 	<-writingDone
+	buf.Flush()
+
+	// The backup file is ready. Run the migration.
+	for _, prefix := range migrationPrefixes {
+		log.VLog("  - Migrating keys in prefix %s", prefix)
+		cidSwapper := CidSwapper{Prefix: prefix, Store: dstore}
+		total, err := cidSwapper.Run(false) // NOT a Dry Run
+		if err != nil {
+			log.Error(err)
+			return err
+		}
+		log.Log("%d CIDv1 keys in %s have been migrated", total, prefix)
+	}
 
 	if err := repo.WriteVersion("9"); err != nil {
 		log.Error("failed to write version file")
@@ -198,18 +211,26 @@ func (m Migration) Revert(opts migrate.Options) error {
 
 	unswapCh := make(chan Swap, 1000)
 	scanner := bufio.NewScanner(f)
+	var scannerErr error
 
 	go func() {
 		defer close(unswapCh)
 
 		for scanner.Scan() {
-			line := scanner.Text()
-			oldAndNew := strings.Split(line, ",")
-			if len(oldAndNew) != 2 {
-				log.Error("bad line in backup file: %s", line)
-				continue
+			cidPath := ds.NewKey(scanner.Text())
+			cidKey := ds.NewKey(cidPath.BaseNamespace())
+			prefix := cidPath.Parent()
+			cid, err := dsKeyToCid(cidKey)
+			if err != nil {
+				log.Error("could not parse cid from backup file: %s", err)
+				scannerErr = err
+				break
 			}
-			sw := Swap{Old: ds.NewKey(oldAndNew[0]), New: ds.NewKey(oldAndNew[1])}
+			mhashPath := prefix.Child(dshelp.MultihashToDsKey(cid.Hash()))
+			// This is the original swap object which is what we
+			// wanted to rebuild. Old is the old path and new is
+			// the new path and the unswapper will revert this.
+			sw := Swap{Old: cidPath, New: mhashPath}
 			unswapCh <- sw
 		}
 		if err := scanner.Err(); err != nil {
@@ -219,13 +240,19 @@ func (m Migration) Revert(opts migrate.Options) error {
 
 	}()
 
-	// The backup file contains prefixed keys, so we do not need to set them.
+	// The backup file contains prefixed keys, so we do not need to set
+	// them.
 	cidSwapper := CidSwapper{Store: dstore}
 	total, err := cidSwapper.Revert(unswapCh)
 	if err != nil {
 		log.Error(err)
 		return err
 	}
+	// Revert will only return after unswapCh is closed, so we know
+	// scannerErr is safe to read at this point.
+	if scannerErr != nil {
+		return err
+	}
 
 	log.Log("%d multihashes reverted to CidV1s", total)
 	if err := repo.WriteVersion("8"); err != nil {

ipfs-8-to-9/migration/swapper.go

@@ -37,10 +37,11 @@ type CidSwapper struct {
 
 // Run lists all the keys in the datastore and triggers a swap operation for
 // those corresponding to CIDv1s (replacing them by their raw multihash).
+// When dryRun is true, it will not perform any changes, but notify SwapCh
+// as if it would.
 //
 // Run returns the total number of keys swapped.
-// The SwapCh is closed at the end of the run.
-func (cswap *CidSwapper) Run() (uint64, error) {
+func (cswap *CidSwapper) Run(dryRun bool) (uint64, error) {
 	// Query all keys. We will loop all keys
 	// and swap those that can be parsed as CIDv1.
 	queryAll := query.Query{
@@ -55,7 +56,7 @@ func (cswap *CidSwapper) Run() (uint64, error) {
 	defer results.Close()
 	resultsCh := results.Next()
 	swapWorkerFunc := func() (uint64, uint64) {
-		return cswap.swapWorker(resultsCh)
+		return cswap.swapWorker(dryRun, resultsCh)
 	}
 	return cswap.runWorkers(NWorkers, swapWorkerFunc)
 }
@@ -100,12 +101,11 @@ func (cswap *CidSwapper) runWorkers(nWorkers int, f func() (uint64, uint64)) (ui
 // swapWorkers reads query results from a channel and renames CIDv1 keys to
 // raw multihashes by reading the blocks and storing them with the new
 // key. Returns the number of keys swapped and the number of errors.
-func (cswap *CidSwapper) swapWorker(resultsCh <-chan query.Result) (uint64, uint64) {
+func (cswap *CidSwapper) swapWorker(dryRun bool, resultsCh <-chan query.Result) (uint64, uint64) {
 	var errored uint64
 
 	sw := &swapWorker{
 		store:      cswap.Store,
-		swapCh:     cswap.SwapCh,
 		syncPrefix: cswap.Prefix,
 	}
 
@@ -133,19 +133,34 @@ func (cswap *CidSwapper) swapWorker(resultsCh <-chan query.Result) (uint64, uint
 		mh := c.Hash()
 		// /path/to/old/<cid> -> /path/to/old/<multihash>
 		newKey := oldKey.Parent().Child(dshelp.MultihashToDsKey(mh))
-		err = sw.swap(oldKey, newKey)
-		if err != nil {
-			log.Error("swapping %s for %s: %s", oldKey, newKey, err)
-			errored++
-			continue
+		if dryRun {
+			sw.swapped++
+		} else {
+			err = sw.swap(oldKey, newKey)
+			if err != nil {
+				log.Error("swapping %s for %s: %s", oldKey, newKey, err)
+				errored++
+				continue
+			}
+		}
+
+		if cswap.SwapCh != nil {
+			cswap.SwapCh <- Swap{Old: oldKey, New: newKey}
 		}
 	}
 
-	// final sync
-	err := sw.sync()
-	if err != nil {
-		log.Error("error performing last sync: %s", err)
-		errored++
+	if !dryRun {
+		// final sync
+		err := sw.syncAndDelete()
+		if err != nil {
+			log.Error("error performing last sync: %s", err)
+			errored++
+		}
+		err = sw.sync() // sync deleted items
+		if err != nil {
+			log.Error("error performing last sync for deletions: %s", err)
+			errored++
+		}
 	}
 
 	return sw.swapped, errored
@@ -159,7 +174,6 @@ func (cswap *CidSwapper) unswapWorker(unswapCh <-chan Swap) (uint64, uint64) {
 
 	swker := &swapWorker{
 		store:      cswap.Store,
-		swapCh:     cswap.SwapCh,
 		syncPrefix: cswap.Prefix,
 	}
 
@@ -194,23 +208,29 @@ func (cswap *CidSwapper) unswapWorker(unswapCh <-chan Swap) (uint64, uint64) {
 				errored++
 			}
 			swker.swapped++
-			continue
-		}
-		if err != nil {
+		} else if err != nil {
 			log.Error("swapping %s for %s: %s", sw.New, sw.Old, err)
 			errored++
 			continue
 		}
+		if cswap.SwapCh != nil {
+			cswap.SwapCh <- Swap{Old: sw.New, New: sw.Old}
+		}
 		// Remember that we switched certain multiash for a Cid already
 		unswappedMap[sw.New] = sw.Old
 	}
 
-	// final sync
-	err := swker.sync()
+	// final sync to added things
+	err := swker.syncAndDelete()
 	if err != nil {
 		log.Error("error performing last sync: %s", err)
 		errored++
 	}
+	err = swker.sync() // final sync for deletes.
+	if err != nil {
+		log.Error("error performing last sync for deletions: %s", err)
+		errored++
+	}
 
 	return swker.swapped, errored
 }
@@ -221,7 +241,6 @@ type swapWorker struct {
 	swapped     uint64
 	curSyncSize uint64
 
-	swapCh     chan Swap
 	store      ds.Batching
 	syncPrefix ds.Key
 
@@ -245,27 +264,20 @@ func (sw *swapWorker) swap(old, new ds.Key) error {
 	sw.swapped++
 	sw.curSyncSize += vLen
 
-	if sw.swapCh != nil {
-		sw.swapCh <- Swap{Old: old, New: new}
-	}
-
 	// We have copied about 10MB
 	if sw.curSyncSize >= SyncSize {
 		sw.curSyncSize = 0
-		err = sw.sync()
+		err = sw.syncAndDelete()
 		if err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
-func (sw *swapWorker) sync() error {
-	log.Log("Syncing after %d objects migrated", sw.swapped)
-	// Sync all the new keys to disk
-	err := sw.store.Sync(sw.syncPrefix)
+func (sw *swapWorker) syncAndDelete() error {
+	err := sw.sync()
 	if err != nil {
-		log.Error(err)
 		return err
 	}
 
@@ -276,11 +288,13 @@ func (sw *swapWorker) sync() error {
 		}
 	}
 	sw.toDelete = nil
+	return nil
+}
 
-	// Sync again.
-	err = sw.store.Sync(sw.syncPrefix)
+func (sw *swapWorker) sync() error {
+	log.Log("Migration worker syncing after %d objects migrated", sw.swapped)
+	err := sw.store.Sync(sw.syncPrefix)
 	if err != nil {
-		log.Error(err)
 		return err
 	}
 	return nil