test: verifyWorkerRun and helptext (#11063)

Author: lidel

core/commands/repo.go

@@ -5,20 +5,22 @@ import (
 	"errors"
 	"fmt"
 	"io"
-	"os"
 	"runtime"
 	"strings"
 	"sync"
 	"text/tabwriter"
+	"time"
 
 	oldcmds "github.com/ipfs/kubo/commands"
 	cmdenv "github.com/ipfs/kubo/core/commands/cmdenv"
+	coreiface "github.com/ipfs/kubo/core/coreiface"
 	corerepo "github.com/ipfs/kubo/core/corerepo"
 	fsrepo "github.com/ipfs/kubo/repo/fsrepo"
 	"github.com/ipfs/kubo/repo/fsrepo/migrations"
 
 	humanize "github.com/dustin/go-humanize"
 	bstore "github.com/ipfs/boxo/blockstore"
+	"github.com/ipfs/boxo/path"
 	cid "github.com/ipfs/go-cid"
 	cmds "github.com/ipfs/go-ipfs-cmds"
 )
@@ -226,45 +228,137 @@ Version         string The repo version.
 	},
 }
 
+// VerifyProgress reports verification progress to the user.
+// It contains either a message about a corrupt block or a progress counter.
 type VerifyProgress struct {
-	Msg      string
-	Progress int
+	Msg      string // Message about a corrupt/healed block (empty for valid blocks)
+	Progress int    // Number of blocks processed so far
 }
 
-func verifyWorkerRun(ctx context.Context, wg *sync.WaitGroup, keys <-chan cid.Cid, results chan<- string, bs bstore.Blockstore) {
+// verifyState represents the state of a block after verification.
+// States track both the verification result and any remediation actions taken.
+type verifyState int
+
+const (
+	verifyStateValid               verifyState = iota // Block is valid and uncorrupted
+	verifyStateCorrupt                                // Block is corrupt, no action taken
+	verifyStateCorruptRemoved                         // Block was corrupt and successfully removed
+	verifyStateCorruptRemoveFailed                    // Block was corrupt but removal failed
+	verifyStateCorruptHealed                          // Block was corrupt, removed, and successfully re-fetched
+	verifyStateCorruptHealFailed                      // Block was corrupt and removed, but re-fetching failed
+)
+
+const (
+	// verifyWorkerMultiplier determines worker pool size relative to CPU count.
+	// Since block verification is I/O-bound (disk reads + potential network fetches),
+	// we use more workers than CPU cores to maximize throughput.
+	verifyWorkerMultiplier = 2
+)
+
+// verifyResult contains the outcome of verifying a single block.
+// It includes the block's CID, its verification state, and an optional
+// human-readable message describing what happened.
+type verifyResult struct {
+	cid   cid.Cid     // CID of the block that was verified
+	state verifyState // Final state after verification and any remediation
+	msg   string      // Human-readable message (empty for valid blocks)
+}
+
+// verifyWorkerRun processes CIDs from the keys channel, verifying their integrity.
+// If shouldDrop is true, corrupt blocks are removed from the blockstore.
+// If shouldHeal is true (implies shouldDrop), removed blocks are re-fetched from the network.
+// The api parameter must be non-nil when shouldHeal is true.
+// healTimeout specifies the maximum time to wait for each block heal (0 = no timeout).
+func verifyWorkerRun(ctx context.Context, wg *sync.WaitGroup, keys <-chan cid.Cid, results chan<- *verifyResult, bs bstore.Blockstore, api coreiface.CoreAPI, shouldDrop, shouldHeal bool, healTimeout time.Duration) {
 	defer wg.Done()
 
+	sendResult := func(r *verifyResult) bool {
+		select {
+		case results <- r:
+			return true
+		case <-ctx.Done():
+			return false
+		}
+	}
+
 	for k := range keys {
 		_, err := bs.Get(ctx, k)
 		if err != nil {
-			select {
-			case results <- fmt.Sprintf("block %s was corrupt (%s)", k, err):
-			case <-ctx.Done():
-				return
+			// Block is corrupt
+			result := &verifyResult{cid: k, state: verifyStateCorrupt}
+
+			if !shouldDrop {
+				result.msg = fmt.Sprintf("block %s was corrupt (%s)", k, err)
+				if !sendResult(result) {
+					return
+				}
+				continue
+			}
+
+			// Try to delete
+			if delErr := bs.DeleteBlock(ctx, k); delErr != nil {
+				result.state = verifyStateCorruptRemoveFailed
+				result.msg = fmt.Sprintf("block %s was corrupt (%s), failed to remove (%s)", k, err, delErr)
+				if !sendResult(result) {
+					return
+				}
+				continue
+			}
+
+			if !shouldHeal {
+				result.state = verifyStateCorruptRemoved
+				result.msg = fmt.Sprintf("block %s was corrupt (%s), removed", k, err)
+				if !sendResult(result) {
+					return
+				}
+				continue
 			}
 
+			// Try to heal by re-fetching from network (api is guaranteed non-nil here)
+			healCtx := ctx
+			var healCancel context.CancelFunc
+			if healTimeout > 0 {
+				healCtx, healCancel = context.WithTimeout(ctx, healTimeout)
+			}
+
+			if _, healErr := api.Block().Get(healCtx, path.FromCid(k)); healErr != nil {
+				result.state = verifyStateCorruptHealFailed
+				result.msg = fmt.Sprintf("block %s was corrupt (%s), removed, failed to heal (%s)", k, err, healErr)
+			} else {
+				result.state = verifyStateCorruptHealed
+				result.msg = fmt.Sprintf("block %s was corrupt (%s), removed, healed", k, err)
+			}
+
+			if healCancel != nil {
+				healCancel()
+			}
+
+			if !sendResult(result) {
+				return
+			}
 			continue
 		}
 
-		select {
-		case results <- "":
-		case <-ctx.Done():
+		// Block is valid
+		if !sendResult(&verifyResult{cid: k, state: verifyStateValid}) {
 			return
 		}
 	}
 }
 
-func verifyResultChan(ctx context.Context, keys <-chan cid.Cid, bs bstore.Blockstore) <-chan string {
-	results := make(chan string)
+// verifyResultChan creates a channel of verification results by spawning multiple worker goroutines
+// to process blocks in parallel. It returns immediately with a channel that will receive results.
+func verifyResultChan(ctx context.Context, keys <-chan cid.Cid, bs bstore.Blockstore, api coreiface.CoreAPI, shouldDrop, shouldHeal bool, healTimeout time.Duration) <-chan *verifyResult {
+	results := make(chan *verifyResult)
 
 	go func() {
 		defer close(results)
 
 		var wg sync.WaitGroup
 
-		for i := 0; i < runtime.NumCPU()*2; i++ {
+		for i := 0; i < runtime.NumCPU()*verifyWorkerMultiplier; i++ {
 			wg.Add(1)
-			go verifyWorkerRun(ctx, &wg, keys, results, bs)
+			go verifyWorkerRun(ctx, &wg, keys, results, bs, api, shouldDrop, shouldHeal, healTimeout)
 		}
 
 		wg.Wait()
@@ -276,13 +370,84 @@ func verifyResultChan(ctx context.Context, keys <-chan cid.Cid, bs bstore.Blocks
 var repoVerifyCmd = &cmds.Command{
 	Helptext: cmds.HelpText{
 		Tagline: "Verify all blocks in repo are not corrupted.",
+		ShortDescription: `
+'ipfs repo verify' checks integrity of all blocks in the local datastore.
+Each block is read and validated against its CID to ensure data integrity.
+
+Without any flags, this is a SAFE, read-only check that only reports corrupt
+blocks without modifying the repository. This can be used as a "dry run" to
+preview what --drop or --heal would do.
+
+Use --drop to remove corrupt blocks, or --heal to remove and re-fetch from
+the network.
+
+Examples:
+  ipfs repo verify          # safe read-only check, reports corrupt blocks
+  ipfs repo verify --drop   # remove corrupt blocks
+  ipfs repo verify --heal   # remove and re-fetch corrupt blocks
+
+Exit Codes:
+  0: All blocks are valid, OR all corrupt blocks were successfully remediated
+     (with --drop or --heal)
+  1: Corrupt blocks detected (without flags), OR remediation failed (block
+     removal or healing failed with --drop or --heal)
+
+Note: --heal requires the daemon to be running in online mode with network
+connectivity to nodes that have the missing blocks. Make sure the daemon is
+online and connected to other peers. Healing will attempt to re-fetch each
+corrupt block from the network after removing it. If a block cannot be found
+on the network, it will remain deleted.
+
+WARNING: Both --drop and --heal are DESTRUCTIVE operations that permanently
+delete corrupt blocks from your repository. Once deleted, blocks cannot be
+recovered unless --heal successfully fetches them from the network. Blocks
+that cannot be healed will remain permanently deleted. Always backup your
+repository before using these options.
+`,
+	},
+	Options: []cmds.Option{
+		cmds.BoolOption("drop", "Remove corrupt blocks from datastore (destructive operation)."),
+		cmds.BoolOption("heal", "Remove corrupt blocks and re-fetch from network (destructive operation, implies --drop)."),
+		cmds.StringOption("heal-timeout", "Maximum time to wait for each block heal (e.g., \"30s\"). Only applies with --heal.").WithDefault("30s"),
 	},
 	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
 		nd, err := cmdenv.GetNode(env)
 		if err != nil {
 			return err
 		}
 
+		drop, _ := req.Options["drop"].(bool)
+		heal, _ := req.Options["heal"].(bool)
+
+		if heal {
+			drop = true // heal implies drop
+		}
+
+		// Parse and validate heal-timeout
+		timeoutStr, _ := req.Options["heal-timeout"].(string)
+		healTimeout, err := time.ParseDuration(timeoutStr)
+		if err != nil {
+			return fmt.Errorf("invalid heal-timeout: %w", err)
+		}
+		if healTimeout < 0 {
+			return errors.New("heal-timeout must be >= 0")
+		}
+
+		// Check online mode and API availability for healing operation
+		var api coreiface.CoreAPI
+		if heal {
+			if !nd.IsOnline {
+				return ErrNotOnline
+			}
+			api, err = cmdenv.GetApi(env, req)
+			if err != nil {
+				return err
+			}
+			if api == nil {
+				return fmt.Errorf("healing requested but API is not available - make sure daemon is online and connected to other peers")
+			}
+		}
+
 		bs := &bstore.ValidatingBlockstore{Blockstore: bstore.NewBlockstore(nd.Repo.Datastore())}
 
 		keys, err := bs.AllKeysChan(req.Context)
@@ -291,17 +456,47 @@ var repoVerifyCmd = &cmds.Command{
 			return err
 		}
 
-		results := verifyResultChan(req.Context, keys, bs)
+		results := verifyResultChan(req.Context, keys, bs, api, drop, heal, healTimeout)
 
-		var fails int
+		// Track statistics for each type of outcome
+		var corrupted, removed, removeFailed, healed, healFailed int
 		var i int
-		for msg := range results {
-			if msg != "" {
-				if err := res.Emit(&VerifyProgress{Msg: msg}); err != nil {
+
+		for result := range results {
+			// Update counters based on the block's final state
+			switch result.state {
+			case verifyStateCorrupt:
+				// Block is corrupt but no action was taken (--drop not specified)
+				corrupted++
+			case verifyStateCorruptRemoved:
+				// Block was corrupt and successfully removed (--drop specified)
+				corrupted++
+				removed++
+			case verifyStateCorruptRemoveFailed:
+				// Block was corrupt but couldn't be removed
+				corrupted++
+				removeFailed++
+			case verifyStateCorruptHealed:
+				// Block was corrupt, removed, and successfully re-fetched (--heal specified)
+				corrupted++
+				removed++
+				healed++
+			case verifyStateCorruptHealFailed:
+				// Block was corrupt and removed, but re-fetching failed
+				corrupted++
+				removed++
+				healFailed++
+			default:
+				// verifyStateValid blocks are not counted (they're the expected case)
+			}
+
+			// Emit progress message for corrupt blocks
+			if result.state != verifyStateValid && result.msg != "" {
+				if err := res.Emit(&VerifyProgress{Msg: result.msg}); err != nil {
 					return err
 				}
-				fails++
 			}
+
 			i++
 			if err := res.Emit(&VerifyProgress{Progress: i}); err != nil {
 				return err
@@ -312,8 +507,42 @@ var repoVerifyCmd = &cmds.Command{
 			return err
 		}
 
-		if fails != 0 {
-			return errors.New("verify complete, some blocks were corrupt")
+		if corrupted > 0 {
+			// Build a summary of what happened with corrupt blocks
+			summary := fmt.Sprintf("verify complete, %d blocks corrupt", corrupted)
+			if removed > 0 {
+				summary += fmt.Sprintf(", %d removed", removed)
+			}
+			if removeFailed > 0 {
+				summary += fmt.Sprintf(", %d failed to remove", removeFailed)
+			}
+			if healed > 0 {
+				summary += fmt.Sprintf(", %d healed", healed)
+			}
+			if healFailed > 0 {
+				summary += fmt.Sprintf(", %d failed to heal", healFailed)
+			}
+
+			// Determine success/failure based on operation mode
+			shouldFail := false
+
+			if !drop {
+				// Detection-only mode: always fail if corruption found
+				shouldFail = true
+			} else if heal {
+				// Heal mode: fail if any removal or heal failed
+				shouldFail = (removeFailed > 0 || healFailed > 0)
+			} else {
+				// Drop mode: fail if any removal failed
+				shouldFail = (removeFailed > 0)
+			}
+
+			if shouldFail {
+				return errors.New(summary)
+			}
+
+			// Success: emit summary as a message instead of error
+			return res.Emit(&VerifyProgress{Msg: summary})
 		}
 
 		return res.Emit(&VerifyProgress{Msg: "verify complete, all blocks validated."})
@@ -322,7 +551,7 @@ var repoVerifyCmd = &cmds.Command{
 	Encoders: cmds.EncoderMap{
 		cmds.Text: cmds.MakeTypedEncoder(func(req *cmds.Request, w io.Writer, obj *VerifyProgress) error {
 			if strings.Contains(obj.Msg, "was corrupt") {
-				fmt.Fprintln(os.Stdout, obj.Msg)
+				fmt.Fprintln(w, obj.Msg)
 				return nil
 			}