add repo verify command

core/commands/repo.go

@@ -3,15 +3,19 @@ package commands
 import (
 	"bytes"
 	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+
+	bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
 	cmds "github.com/ipfs/go-ipfs/commands"
 	corerepo "github.com/ipfs/go-ipfs/core/corerepo"
 	config "github.com/ipfs/go-ipfs/repo/config"
 	fsrepo "github.com/ipfs/go-ipfs/repo/fsrepo"
 	lockfile "github.com/ipfs/go-ipfs/repo/fsrepo/lock"
+
 	u "gx/ipfs/QmZNVWh8LLjAavuQ2JXuFmuYH3C11xo988vSgp7UQrTRj1/go-ipfs-util"
-	"io"
-	"os"
-	"path/filepath"
 )
 
 type RepoVersion struct {
@@ -31,6 +35,7 @@ var RepoCmd = &cmds.Command{
 		"stat":    repoStatCmd,
 		"fsck":    RepoFsckCmd,
 		"version": repoVersionCmd,
+		"verify":  repoVerifyCmd,
 	},
 }
 
@@ -207,16 +212,100 @@ daemons are running.
 			return
 		}
 
-		s := "Lockfiles have been removed."
-		log.Info(s)
-		res.SetOutput(&MessageOutput{s + "\n"})
+		res.SetOutput(&MessageOutput{"Lockfiles have been removed.\n"})
 	},
 	Type: MessageOutput{},
 	Marshalers: cmds.MarshalerMap{
 		cmds.Text: MessageTextMarshaler,
 	},
 }
 
+type VerifyProgress struct {
+	Message  string
+	Progress int
+}
+
+var repoVerifyCmd = &cmds.Command{
+	Helptext: cmds.HelpText{
+		Tagline: "Verify all blocks in repo are not corrupted.",
+	},
+	Run: func(req cmds.Request, res cmds.Response) {
+		nd, err := req.InvocContext().GetNode()
+		if err != nil {
+			res.SetError(err, cmds.ErrNormal)
+			return
+		}
+
+		out := make(chan interface{})
+		go func() {
+			defer close(out)
+			bs := bstore.NewBlockstore(nd.Repo.Datastore())
+
+			bs.RuntimeHashing(true)
+
+			keys, err := bs.AllKeysChan(req.Context())
+			if err != nil {
+				log.Error(err)
+				return
+			}
+
+			var fails int
+			var i int
+			for k := range keys {
+				_, err := bs.Get(k)
+				if err != nil {
+					out <- &VerifyProgress{
+						Message: fmt.Sprintf("block %s was corrupt (%s)", k, err),
+					}
+					fails++
+				}
+				i++
+				out <- &VerifyProgress{Progress: i}
+			}
+			if fails == 0 {
+				out <- &VerifyProgress{Message: "verify complete, all blocks validated."}
+			} else {
+				out <- &VerifyProgress{Message: "verify complete, some blocks were corrupt."}
+			}
+		}()
+
+		res.SetOutput((<-chan interface{})(out))
+	},
+	Marshalers: cmds.MarshalerMap{
+		cmds.Text: func(res cmds.Response) (io.Reader, error) {
+			out := res.Output().(<-chan interface{})
+
+			marshal := func(v interface{}) (io.Reader, error) {
+				obj, ok := v.(*VerifyProgress)
+				if !ok {
+					return nil, u.ErrCast()
+				}
+
+				buf := new(bytes.Buffer)
+				if obj.Message != "" {
+					if strings.Contains(obj.Message, "blocks were corrupt") {
+						return nil, fmt.Errorf(obj.Message)
+					}
+					if len(obj.Message) < 20 {
+						obj.Message += "             "
+					}
+					fmt.Fprintln(buf, obj.Message)
+					return buf, nil
+				}
+
+				fmt.Fprintf(buf, "%d blocks processed.\r", obj.Progress)
+				return buf, nil
+			}
+
+			return &cmds.ChannelMarshaler{
+				Channel:   out,
+				Marshaler: marshal,
+				Res:       res,
+			}, nil
+		},
+	},
+}
+
 var repoVersionCmd = &cmds.Command{
 	Helptext: cmds.HelpText{
 		Tagline: "Show the repo version.",

test/sharness/t0084-repo-read-rehash.sh

@@ -34,4 +34,9 @@ test_expect_success 'getting modified block fails' '
 	grep "block in storage has different hash than requested" err_msg
 '
 
+test_expect_success "block shows up in repo verify" '
+	test_expect_code 1 ipfs repo verify > verify_out &&
+	grep "$H_BLOCK2" verify_out
+'
+
 test_done