Merge pull request #150 from ipfs/fix/11-to-12-querying

11-to-12: Re-work logic to use the backup file for both migrate and revert

fs-repo-11-to-12/go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/ipfs/fs-repo-migrations/tools v0.0.0-20211209222258-754a2dcb82ea
 	github.com/ipfs/go-cid v0.0.7
 	github.com/ipfs/go-datastore v0.4.5
-	github.com/ipfs/go-ds-badger v0.2.7-0.20211210151007-a2805355dcf5 // indirect
+	github.com/ipfs/go-ds-badger v0.2.7-0.20220117180822-159330558612 // indirect
 	github.com/ipfs/go-filestore v1.0.0
 	github.com/ipfs/go-ipfs v0.8.0
 	github.com/ipfs/go-ipfs-ds-help v1.0.0

fs-repo-11-to-12/go.sum

@@ -295,8 +295,8 @@ github.com/ipfs/go-ds-badger v0.0.7/go.mod h1:qt0/fWzZDoPW6jpQeqUjR5kBfhDNB65jd9
 github.com/ipfs/go-ds-badger v0.2.1/go.mod h1:Tx7l3aTph3FMFrRS838dcSJh+jjA7cX9DrGVwx/NOwE=
 github.com/ipfs/go-ds-badger v0.2.3/go.mod h1:pEYw0rgg3FIrywKKnL+Snr+w/LjJZVMTBRn4FS6UHUk=
 github.com/ipfs/go-ds-badger v0.2.6/go.mod h1:02rnztVKA4aZwDuaRPTf8mpqcKmXP7mLl6JPxd14JHA=
-github.com/ipfs/go-ds-badger v0.2.7-0.20211210151007-a2805355dcf5 h1:ovdpQk2ZVK6eQLzMCZy1z2tJae7yvE8xaPUw4Pr1RqI=
-github.com/ipfs/go-ds-badger v0.2.7-0.20211210151007-a2805355dcf5/go.mod h1:02rnztVKA4aZwDuaRPTf8mpqcKmXP7mLl6JPxd14JHA=
+github.com/ipfs/go-ds-badger v0.2.7-0.20220117180822-159330558612 h1:Uvp2/ZNlR3YmH04XJGv4YsPabhPzRPyeroLltVKBSr8=
+github.com/ipfs/go-ds-badger v0.2.7-0.20220117180822-159330558612/go.mod h1:02rnztVKA4aZwDuaRPTf8mpqcKmXP7mLl6JPxd14JHA=
 github.com/ipfs/go-ds-flatfs v0.4.5 h1:4QceuKEbH+HVZ2ZommstJMi3o3II+dWS3IhLaD7IGHs=
 github.com/ipfs/go-ds-flatfs v0.4.5/go.mod h1:e4TesLyZoA8k1gV/yCuBTnt2PJtypn4XUlB5n8KQMZY=
 github.com/ipfs/go-ds-leveldb v0.0.1/go.mod h1:feO8V3kubwsEF22n0YRQCffeb79OOYIykR4L04tMOYc=

fs-repo-11-to-12/migration/migration.go

@@ -92,7 +92,6 @@ func (m *Migration) Apply(opts migrate.Options) error {
 		log.Error(err)
 		return err
 	}
-	defer f.Close()
 	buf := bufio.NewWriter(f)
 
 	swapCh := make(chan Swap, 1000)
@@ -119,7 +118,7 @@ func (m *Migration) Apply(opts migrate.Options) error {
 	for _, prefix := range migrationPrefixes {
 		log.VLog("  - Adding keys in prefix %s to backup file", prefix)
 		cidSwapper := CidSwapper{Prefix: prefix, Store: m.dstore, SwapCh: swapCh}
-		total, err := cidSwapper.Run(true) // DRY RUN
+		total, err := cidSwapper.Prepare() // DRY RUN
 		if err != nil {
 			close(swapCh)
 			log.Error(err)
@@ -131,17 +130,12 @@ func (m *Migration) Apply(opts migrate.Options) error {
 	// Wait for our writing to finish before doing the flushing.
 	<-writingDone
 	buf.Flush()
+	f.Close()
 
-	// MIGRATION: Run the real migration.
-	for _, prefix := range migrationPrefixes {
-		log.VLog("  - Migrating keys in prefix %s", prefix)
-		cidSwapper := CidSwapper{Prefix: prefix, Store: m.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)
+	err = m.scanAndSwap(filepath.Join(opts.Path, backupFile), false) // revert=false
+	if err != nil {
+		log.Error(err)
+		return err
 	}
 
 	// Wrap up, we are now in repo-version 12.
@@ -187,21 +181,48 @@ func (m *Migration) Revert(opts migrate.Options) error {
 
 	// Open revert path for reading
 	backupPath := filepath.Join(opts.Path, backupFile)
+	err = m.scanAndSwap(backupPath, true) // revert = true
+	if err != nil {
+		log.Error(err)
+		return err
+	}
+
+	// Wrap up the Revert. We are back at version 11.
+	if err := repo.WriteVersion("11"); err != nil {
+		log.Error("failed to write version file")
+		return err
+	}
+
+	log.Log("reverted version file to version 11")
+
+	// Move the backup file out of the way.
+	err = os.Rename(backupPath, backupPath+".reverted")
+	if err != nil {
+		log.Error("could not rename the backup file, but migration worked: %s", err)
+		return err
+	}
+	return nil
+}
+
+// Receives a backup file which contains all the things that need to be
+// migrated and reads every line, performing swaps in the needed direction.
+func (m *Migration) scanAndSwap(backupPath string, revert bool) error {
 	f, err := getBackupFile(backupPath)
 	if err != nil {
 		log.Error(err)
 		return err
 	}
+	defer f.Close()
 
-	unswapCh := make(chan Swap, 1000)
+	swapCh := make(chan Swap, 1000)
 	scanner := bufio.NewScanner(f)
 	var scannerErr error
 
-	// This will send swap objects to the Unswapper on unswapCh as they
+	// This will send swap objects to the swapping channel as they
 	// are read from the backup file on disk. It will also send MFS and
-	// pinset pins for reversal.
+	// pinset pins for reversal when doing a revert.
 	go func() {
-		defer close(unswapCh)
+		defer close(swapCh)
 
 		// Process backup file first.
 		for scanner.Scan() {
@@ -222,60 +243,53 @@ func (m *Migration) Revert(opts migrate.Options) error {
 				break
 			}
 			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.
+
+			// The swapper will move cidPath to mhashPath, and the unswapper
+			// will do the opposite.
 			sw := Swap{Old: cidPath, New: mhashPath}
-			unswapCh <- sw
+			swapCh <- sw
 		}
 		if err := scanner.Err(); err != nil {
 			log.Error(err)
 			return
 		}
 
-		// Process MFS/pinset. We have to do this in cases the user
-		// has been running with the migration for some time and made changes to
-		// the pinset or the MFS root.
-		if err := walkPinsAndMFS(unswapCh, m.dstore); err != nil {
-			log.Error(err)
-			return
+		if revert {
+			// Process MFS/pinset. We have to do this in cases the
+			// user has been running with the migration for some
+			// time and made changes to the pinset or the MFS
+			// root.
+			if err := walkPinsAndMFS(swapCh, m.dstore); err != nil {
+				log.Error(err)
+				return
+			}
 		}
-
 	}()
 
 	// The backup file contains prefixed keys, so we do not need to set
 	// Prefix in the CidSwapper.
 	cidSwapper := CidSwapper{Store: m.dstore}
-	total, err := cidSwapper.Revert(unswapCh)
+	var total uint64
+	if revert {
+		total, err = cidSwapper.Revert(swapCh)
+	} else {
+		total, err = cidSwapper.Run(swapCh)
+	}
 	if err != nil {
 		log.Error(err)
 		return err
 	}
-	// Revert will only return after unswapCh is closed, so we know
+
+	// The swapper will only return after swapCh is closed, so we know
 	// scannerErr is safe to read at this point.
 	if scannerErr != nil {
 		return err
 	}
 
-	// Wrap up the Revert. We are back at version 11.
-	log.Log("%d multihashes reverted to CidV1s", total)
-	if err := repo.WriteVersion("11"); err != nil {
-		log.Error("failed to write version file")
-		return err
-	}
-
-	log.Log("reverted version file to version 11")
-	err = f.Close()
-	if err != nil {
-		log.Error("could not close backup file")
-		return err
-	}
-
-	// Move the backup file out of the way.
-	err = os.Rename(backupPath, backupPath+".reverted")
-	if err != nil {
-		log.Error("could not rename the backup file, but migration worked: %s", err)
-		return err
+	if revert {
+		log.Log("%d multihashes swapped to CidV1s", total)
+	} else {
+		log.Log("%d CidV1s swapped to multihashes", total)
 	}
 	return nil
 }